功能：增加旧测试数据闭包纳入

shierfeng-fastapi-backend/module_business/dao/test_batch_adoption_dao.py

@@ -0,0 +1,100 @@
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from exceptions.exception import ServiceWarning
+from module_business.domain.test_batch_graph import BATCH_MODELS, ROOT_TABLES, BatchSelection
+from module_business.entity.vo.test_batch_vo import AdoptionRootModel
+
+
+class TestBatchAdoptionDao:
+    """旧数据闭包锁定、展示与批量纳入。"""
+
+    @classmethod
+    async def lock_roots(
+        cls,
+        db: AsyncSession,
+        roots: list[AdoptionRootModel],
+    ) -> None:
+        requested: dict[str, set[int]] = {}
+        for root in roots:
+            requested.setdefault(ROOT_TABLES[root.resource_type], set()).update(root.resource_ids)
+        for table_name in sorted(requested):
+            primary_key = BATCH_MODELS[table_name][1]
+            ids = sorted(requested[table_name])
+            result = await db.execute(
+                select(primary_key).where(primary_key.in_(ids)).order_by(primary_key).with_for_update()
+            )
+            if {row[0] for row in result.all()} != set(ids):
+                raise ServiceWarning('', '指定的旧数据不存在')
+
+    @classmethod
+    async def lock_selection(cls, db: AsyncSession, selection: BatchSelection) -> None:
+        for table_name in sorted(selection):
+            ids = sorted(selection[table_name])
+            if not ids:
+                continue
+            primary_key = BATCH_MODELS[table_name][1]
+            result = await db.execute(
+                select(primary_key).where(primary_key.in_(ids)).order_by(primary_key).with_for_update()
+            )
+            if len(result.all()) != len(ids):
+                raise ServiceWarning('', '预览范围已变化')
+
+    @classmethod
+    async def find_tagged_record(
+        cls,
+        db: AsyncSession,
+        selection: BatchSelection,
+    ) -> tuple[str, int] | None:
+        for table_name in sorted(selection):
+            ids = sorted(selection[table_name])
+            if not ids:
+                continue
+            model, primary_key, _ = BATCH_MODELS[table_name]
+            result = await db.execute(
+                select(primary_key)
+                .where(
+                    primary_key.in_(ids),
+                    model.test_batch_id.is_not(None),
+                )
+                .order_by(primary_key)
+                .limit(1)
+            )
+            record_id = result.scalar_one_or_none()
+            if record_id is not None:
+                return table_name, record_id
+        return None
+
+    @classmethod
+    async def list_display_keys(
+        cls,
+        db: AsyncSession,
+        selection: BatchSelection,
+    ) -> dict[str, list[str]]:
+        keys: dict[str, list[str]] = {}
+        for table_name in BATCH_MODELS:
+            ids = sorted(selection[table_name])
+            if not ids:
+                keys[table_name] = []
+                continue
+            _, primary_key, display_column = BATCH_MODELS[table_name]
+            result = await db.execute(select(display_column).where(primary_key.in_(ids)).order_by(display_column))
+            keys[table_name] = sorted(str(row[0]) for row in result.all())
+        return keys
+
+    @classmethod
+    async def set_test_batch_id(
+        cls,
+        db: AsyncSession,
+        table_name: str,
+        ids: list[int],
+        batch_id: int,
+    ) -> None:
+        if not ids:
+            return
+        model, primary_key, _ = BATCH_MODELS[table_name]
+        result = await db.execute(
+            update(model).where(primary_key.in_(ids), model.test_batch_id.is_(None)).values(test_batch_id=batch_id)
+        )
+        if result.rowcount != len(ids):
+            raise ServiceWarning('', '存在记录已属于其他测试批次')

shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py

@@ -149,8 +149,39 @@ class TestBatchDao:
     async def count_owned_rows(cls, db: AsyncSession, batch_id: int) -> dict[str, int]:
         counts: dict[str, int] = {}
         for model in _BATCH_OWNED_MODELS:
-            result = await db.execute(
+            result = await db.execute(select(func.count()).select_from(model).where(model.test_batch_id == batch_id))
-                select(func.count()).select_from(model).where(model.test_batch_id == batch_id)
-            )
             counts[model.__tablename__] = result.scalar() or 0
         return counts
+
+    @classmethod
+    async def store_preview(
+        cls,
+        db: AsyncSession,
+        batch: BizTestBatch,
+        *,
+        action: str,
+        token_hash: str,
+        scope_hash: str,
+        expires_at: datetime,
+        update_by: str,
+        update_time: datetime,
+    ) -> None:
+        """在已锁定的批次对象上保存一次性预览状态。"""
+        batch.preview_action = action
+        batch.preview_token_hash = token_hash
+        batch.preview_scope_hash = scope_hash
+        batch.preview_expires_at = expires_at
+        batch.update_by = update_by
+        batch.update_time = update_time
+
+    @classmethod
+    async def clear_preview(
+        cls,
+        db: AsyncSession,
+        batch: BizTestBatch,
+    ) -> None:
+        """消费已锁定批次上的预览令牌与范围摘要。"""
+        batch.preview_action = None
+        batch.preview_token_hash = None
+        batch.preview_scope_hash = None
+        batch.preview_expires_at = None

shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py

@@ -0,0 +1,261 @@
+import hashlib
+import json
+import secrets
+from collections import defaultdict, deque
+from dataclasses import dataclass
+
+from sqlalchemy import Column, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config.database import Base
+from module_business.entity.do.master_do import (
+    BizCustomer,
+    BizCustomerSite,
+    BizPallet,
+    BizProductionLine,
+    BizStrain,
+    BizWarehouse,
+)
+from module_business.entity.do.operation_do import (
+    BizFillingRecord,
+    BizInboundItem,
+    BizInboundOrder,
+    BizPalletEvent,
+    BizPalletInventory,
+    BizReturnItem,
+    BizReturnOrder,
+    BizShipmentItem,
+    BizShipmentOrder,
+    BizUsageRecord,
+)
+from module_business.entity.vo.test_batch_vo import AdoptionRootModel
+
+BatchSelection = dict[str, frozenset[int]]
+
+BATCH_MODELS = {
+    'biz_strain': (BizStrain, BizStrain.strain_id, BizStrain.strain_code),
+    'biz_customer': (BizCustomer, BizCustomer.customer_id, BizCustomer.customer_code),
+    'biz_customer_site': (BizCustomerSite, BizCustomerSite.site_id, BizCustomerSite.site_code),
+    'biz_production_line': (
+        BizProductionLine,
+        BizProductionLine.line_id,
+        BizProductionLine.line_code,
+    ),
+    'biz_warehouse': (BizWarehouse, BizWarehouse.warehouse_id, BizWarehouse.warehouse_code),
+    'biz_pallet': (BizPallet, BizPallet.pallet_id, BizPallet.pallet_code),
+    'biz_inbound_order': (BizInboundOrder, BizInboundOrder.inbound_id, BizInboundOrder.inbound_no),
+    'biz_inbound_item': (
+        BizInboundItem,
+        BizInboundItem.inbound_item_id,
+        BizInboundItem.inbound_item_id,
+    ),
+    'biz_filling_record': (
+        BizFillingRecord,
+        BizFillingRecord.filling_id,
+        BizFillingRecord.filling_no,
+    ),
+    'biz_shipment_order': (
+        BizShipmentOrder,
+        BizShipmentOrder.shipment_id,
+        BizShipmentOrder.shipment_no,
+    ),
+    'biz_shipment_item': (
+        BizShipmentItem,
+        BizShipmentItem.shipment_item_id,
+        BizShipmentItem.shipment_item_id,
+    ),
+    'biz_usage_record': (BizUsageRecord, BizUsageRecord.usage_id, BizUsageRecord.usage_no),
+    'biz_return_order': (BizReturnOrder, BizReturnOrder.return_id, BizReturnOrder.return_no),
+    'biz_return_item': (
+        BizReturnItem,
+        BizReturnItem.return_item_id,
+        BizReturnItem.return_item_id,
+    ),
+    'biz_pallet_inventory': (
+        BizPalletInventory,
+        BizPalletInventory.pallet_id,
+        BizPalletInventory.pallet_id,
+    ),
+    'biz_pallet_event': (BizPalletEvent, BizPalletEvent.event_id, BizPalletEvent.event_id),
+}
+
+BATCH_LABELS = {
+    'biz_strain': '菌种',
+    'biz_customer': '客户',
+    'biz_customer_site': '客户现场',
+    'biz_production_line': '生产线',
+    'biz_warehouse': '仓库',
+    'biz_pallet': '托盘',
+    'biz_inbound_order': '入库单',
+    'biz_inbound_item': '入库明细',
+    'biz_filling_record': '装填记录',
+    'biz_shipment_order': '发货单',
+    'biz_shipment_item': '发货明细',
+    'biz_usage_record': '使用记录',
+    'biz_return_order': '退回单',
+    'biz_return_item': '退回明细',
+    'biz_pallet_inventory': '托盘库存',
+    'biz_pallet_event': '托盘事件',
+}
+
+ROOT_TABLES = {
+    'strain': 'biz_strain',
+    'customer': 'biz_customer',
+    'site': 'biz_customer_site',
+    'line': 'biz_production_line',
+    'warehouse': 'biz_warehouse',
+    'pallet': 'biz_pallet',
+    'inbound': 'biz_inbound_order',
+    'filling': 'biz_filling_record',
+    'shipment': 'biz_shipment_order',
+    'usage': 'biz_usage_record',
+    'return': 'biz_return_order',
+}
+
+_EVENT_SOURCE_TABLES = {
+    'INBOUND': 'biz_inbound_order',
+    'FILLING': 'biz_filling_record',
+    'SHIPMENT': 'biz_shipment_order',
+    'USAGE': 'biz_usage_record',
+    'RETURN': 'biz_return_order',
+}
+
+
+def canonical_selection(selection: BatchSelection) -> bytes:
+    payload = {table: sorted(ids) for table, ids in sorted(selection.items()) if ids}
+    return json.dumps(payload, ensure_ascii=True, separators=(',', ':'), sort_keys=True).encode()
+
+
+def selection_hash(selection: BatchSelection) -> str:
+    return hashlib.sha256(canonical_selection(selection)).hexdigest()
+
+
+def issue_preview_token() -> tuple[str, str]:
+    raw = secrets.token_urlsafe(32)
+    return raw, hashlib.sha256(raw.encode()).hexdigest()
+
+
+@dataclass(frozen=True)
+class _GraphEdge:
+    source_table: str
+    target_table: str
+    filter_column: Column
+    result_column: Column
+    source_type: str | None = None
+
+
+class TestBatchGraph:
+    """基于批次模型外键计算可原子纳入的双向关联闭包。"""
+
+    @staticmethod
+    def _foreign_key_edges() -> dict[str, list[_GraphEdge]]:
+        edges: dict[str, list[_GraphEdge]] = defaultdict(list)
+        batch_tables = set(BATCH_MODELS)
+        for child_table_name in sorted(batch_tables):
+            child_table = Base.metadata.tables[child_table_name]
+            child_pk = BATCH_MODELS[child_table_name][1]
+            for foreign_key in sorted(
+                child_table.foreign_keys,
+                key=lambda item: (item.parent.name, item.target_fullname),
+            ):
+                parent_table_name = foreign_key.column.table.name
+                if foreign_key.parent.name == 'test_batch_id' or parent_table_name not in batch_tables:
+                    continue
+                edges[child_table_name].append(
+                    _GraphEdge(
+                        source_table=child_table_name,
+                        target_table=parent_table_name,
+                        filter_column=child_pk,
+                        result_column=foreign_key.parent,
+                    )
+                )
+                edges[parent_table_name].append(
+                    _GraphEdge(
+                        source_table=parent_table_name,
+                        target_table=child_table_name,
+                        filter_column=foreign_key.parent,
+                        result_column=child_pk,
+                    )
+                )
+        return edges
+
+    @classmethod
+    def _all_edges(cls) -> dict[str, list[_GraphEdge]]:
+        edges = cls._foreign_key_edges()
+        event_pk = BATCH_MODELS['biz_pallet_event'][1]
+        for source_type, root_table in sorted(_EVENT_SOURCE_TABLES.items()):
+            edges[root_table].append(
+                _GraphEdge(
+                    source_table=root_table,
+                    target_table='biz_pallet_event',
+                    filter_column=BizPalletEvent.source_id,
+                    result_column=event_pk,
+                    source_type=source_type,
+                )
+            )
+            edges['biz_pallet_event'].append(
+                _GraphEdge(
+                    source_table='biz_pallet_event',
+                    target_table=root_table,
+                    filter_column=event_pk,
+                    result_column=BizPalletEvent.source_id,
+                    source_type=source_type,
+                )
+            )
+        for table_edges in edges.values():
+            table_edges.sort(
+                key=lambda edge: (
+                    edge.target_table,
+                    edge.filter_column.name,
+                    edge.result_column.name,
+                    edge.source_type or '',
+                )
+            )
+        return edges
+
+    @staticmethod
+    async def _existing_ids(db: AsyncSession, table_name: str, ids: set[int]) -> set[int]:
+        if not ids:
+            return set()
+        primary_key = BATCH_MODELS[table_name][1]
+        result = await db.execute(select(primary_key).where(primary_key.in_(sorted(ids))).order_by(primary_key))
+        return {row[0] for row in result.all()}
+
+    @classmethod
+    async def compute_closure(
+        cls,
+        db: AsyncSession,
+        roots: list[AdoptionRootModel],
+    ) -> BatchSelection:
+        selected: dict[str, set[int]] = {table_name: set() for table_name in BATCH_MODELS}
+        requested: dict[str, set[int]] = defaultdict(set)
+        for root in roots:
+            requested[ROOT_TABLES[root.resource_type]].update(root.resource_ids)
+
+        queue: deque[str] = deque()
+        for table_name in sorted(requested):
+            existing = await cls._existing_ids(db, table_name, requested[table_name])
+            selected[table_name].update(existing)
+            if existing:
+                queue.append(table_name)
+
+        expanded: dict[str, set[int]] = {table_name: set() for table_name in BATCH_MODELS}
+        edges = cls._all_edges()
+        while queue:
+            source_table = queue.popleft()
+            pending_ids = selected[source_table] - expanded[source_table]
+            if not pending_ids:
+                continue
+            expanded[source_table].update(pending_ids)
+            for edge in edges.get(source_table, []):
+                statement = select(edge.result_column).where(edge.filter_column.in_(sorted(pending_ids)))
+                if edge.source_type is not None:
+                    statement = statement.where(BizPalletEvent.source_type == edge.source_type)
+                result = await db.execute(statement.order_by(edge.result_column))
+                related_ids = {row[0] for row in result.all() if row[0] is not None}
+                new_ids = related_ids - selected[edge.target_table]
+                if new_ids:
+                    selected[edge.target_table].update(new_ids)
+                    queue.append(edge.target_table)
+
+        return {table_name: frozenset(selected[table_name]) for table_name in BATCH_MODELS}

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

@@ -0,0 +1,167 @@
+import hashlib
+import hmac
+from datetime import datetime, timedelta
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.vo import CrudResponseModel
+from exceptions.exception import PermissionException, ServiceWarning
+from module_admin.entity.vo.user_vo import CurrentUserModel
+from module_business.dao.test_batch_adoption_dao import TestBatchAdoptionDao
+from module_business.dao.test_batch_dao import TestBatchDao
+from module_business.domain.enums import TestBatchStatus
+from module_business.domain.test_batch_graph import (
+    BATCH_LABELS,
+    BATCH_MODELS,
+    TestBatchGraph,
+    issue_preview_token,
+    selection_hash,
+)
+from module_business.entity.vo.test_batch_vo import (
+    AdoptionExecuteRequestModel,
+    AdoptionPreviewRequestModel,
+    BatchImpactModel,
+    TestBatchPreviewModel,
+)
+from module_business.service.test_batch_access_service import TestBatchAccessService
+
+
+class TestBatchAdoptionService:
+    """预览并原子纳入已存在的正式数据闭包。"""
+
+    @staticmethod
+    def _assert_admin(current_user: CurrentUserModel) -> None:
+        TestBatchAccessService.assert_super_admin(current_user)
+
+    @staticmethod
+    def _user_name(current_user: CurrentUserModel) -> str:
+        user = current_user.user
+        if user is None or user.user_name is None:
+            raise PermissionException('', '当前用户信息不可用')
+        return user.user_name
+
+    @staticmethod
+    def _assert_active(batch: Any | None) -> None:
+        if batch is None:
+            raise ServiceWarning('', '测试批次不存在')
+        status = batch.status
+        if status == TestBatchStatus.CLOSED.value:
+            raise ServiceWarning('', '测试批次已关闭')
+        if status == TestBatchStatus.PURGED.value:
+            raise ServiceWarning('', '测试批次已清理')
+        if status != TestBatchStatus.ACTIVE.value:
+            raise ServiceWarning('', '测试批次状态异常')
+
+    @staticmethod
+    def _assert_token(batch: object, raw_token: str, now: datetime) -> None:
+        supplied_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+        stored_hash = getattr(batch, 'preview_token_hash', None)
+        comparable_hash = stored_hash if isinstance(stored_hash, str) else '0' * 64
+        if not hmac.compare_digest(comparable_hash, supplied_hash):
+            raise ServiceWarning('', '预览令牌无效')
+        if getattr(batch, 'preview_action', None) != 'ADOPT':
+            raise ServiceWarning('', '预览令牌无效')
+        expires_at = getattr(batch, 'preview_expires_at', None)
+        if expires_at is None or expires_at <= now:
+            raise ServiceWarning('', '预览令牌已过期')
+
+    @classmethod
+    async def preview(
+        cls,
+        db: AsyncSession,
+        batch_id: int,
+        command: AdoptionPreviewRequestModel,
+        current_user: CurrentUserModel,
+    ) -> TestBatchPreviewModel:
+        cls._assert_admin(current_user)
+        user_name = cls._user_name(current_user)
+        try:
+            batch = await TestBatchDao.get_batch_for_update(db, batch_id)
+            cls._assert_active(batch)
+            selection = await TestBatchGraph.compute_closure(db, command.roots)
+            if await TestBatchAdoptionDao.find_tagged_record(db, selection) is not None:
+                raise ServiceWarning('', '存在记录已属于其他测试批次')
+            display_keys = await TestBatchAdoptionDao.list_display_keys(db, selection)
+
+            now = datetime.now()
+            expires_at = now + timedelta(seconds=600)
+            raw_token, token_hash = issue_preview_token()
+            await TestBatchDao.store_preview(
+                db,
+                batch,
+                action='ADOPT',
+                token_hash=token_hash,
+                scope_hash=selection_hash(selection),
+                expires_at=expires_at,
+                update_by=user_name,
+                update_time=now,
+            )
+            await db.commit()
+            impacts = [
+                BatchImpactModel(
+                    table_name=table_name,
+                    label=BATCH_LABELS[table_name],
+                    count=len(selection[table_name]),
+                    keys=display_keys[table_name],
+                )
+                for table_name in BATCH_MODELS
+            ]
+            return TestBatchPreviewModel(
+                batch_id=batch_id,
+                batch_name=batch.batch_name,
+                preview_token=raw_token,
+                expires_at=expires_at,
+                total_count=sum(item.count for item in impacts),
+                impacts=impacts,
+            )
+        except Exception:
+            await db.rollback()
+            raise
+
+    @classmethod
+    async def execute(
+        cls,
+        db: AsyncSession,
+        batch_id: int,
+        command: AdoptionExecuteRequestModel,
+        current_user: CurrentUserModel,
+    ) -> CrudResponseModel:
+        cls._assert_admin(current_user)
+        user_name = cls._user_name(current_user)
+        try:
+            batch = await TestBatchDao.get_batch_for_update(db, batch_id)
+            cls._assert_active(batch)
+            if command.batch_name != batch.batch_name:
+                raise ServiceWarning('', '测试批次名称不匹配')
+            cls._assert_token(batch, command.preview_token, datetime.now())
+
+            await TestBatchAdoptionDao.lock_roots(db, command.roots)
+            first_selection = await TestBatchGraph.compute_closure(db, command.roots)
+            await TestBatchAdoptionDao.lock_selection(db, first_selection)
+            locked_selection = await TestBatchGraph.compute_closure(db, command.roots)
+            expected_scope_hash = batch.preview_scope_hash
+            if (
+                selection_hash(first_selection) != expected_scope_hash
+                or selection_hash(locked_selection) != expected_scope_hash
+            ):
+                raise ServiceWarning('', '预览范围已变化')
+            if await TestBatchAdoptionDao.find_tagged_record(db, locked_selection) is not None:
+                raise ServiceWarning('', '存在记录已属于其他测试批次')
+
+            for table_name in sorted(locked_selection):
+                await TestBatchAdoptionDao.set_test_batch_id(
+                    db,
+                    table_name,
+                    sorted(locked_selection[table_name]),
+                    batch_id,
+                )
+            now = datetime.now()
+            await TestBatchDao.clear_preview(db, batch)
+            batch.update_by = user_name
+            batch.update_time = now
+            await db.commit()
+            return CrudResponseModel(is_success=True, message='旧测试数据纳入成功')
+        except Exception:
+            await db.rollback()
+            raise

shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py

@@ -0,0 +1,447 @@
+import inspect
+import re
+import unittest
+from contextlib import ExitStack, contextmanager
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+from typing import Any, get_type_hints
+from unittest.mock import AsyncMock, patch
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.vo import CrudResponseModel
+from exceptions.exception import PermissionException, ServiceWarning
+from module_admin.entity.vo.user_vo import CurrentUserModel
+from module_business.dao.test_batch_adoption_dao import TestBatchAdoptionDao
+from module_business.dao.test_batch_dao import TestBatchDao
+from module_business.domain.test_batch_graph import (
+    BATCH_MODELS,
+    BatchSelection,
+    TestBatchGraph,
+    canonical_selection,
+    selection_hash,
+)
+from module_business.entity.vo.test_batch_vo import (
+    AdoptionExecuteRequestModel,
+    AdoptionPreviewRequestModel,
+    AdoptionRootModel,
+    TestBatchPreviewModel,
+)
+from module_business.service.test_batch_adoption_service import TestBatchAdoptionService
+from tests.module_business.test_batch_fixtures import make_admin, make_current_user
+
+TABLE_LABELS = {
+    'biz_strain': '菌种',
+    'biz_customer': '客户',
+    'biz_customer_site': '客户现场',
+    'biz_production_line': '生产线',
+    'biz_warehouse': '仓库',
+    'biz_pallet': '托盘',
+    'biz_inbound_order': '入库单',
+    'biz_inbound_item': '入库明细',
+    'biz_filling_record': '装填记录',
+    'biz_shipment_order': '发货单',
+    'biz_shipment_item': '发货明细',
+    'biz_usage_record': '使用记录',
+    'biz_return_order': '退回单',
+    'biz_return_item': '退回明细',
+    'biz_pallet_inventory': '托盘库存',
+    'biz_pallet_event': '托盘事件',
+}
+
+
+class _RowsResult:
+    def __init__(self, rows: list[tuple[Any, ...]]) -> None:
+        self._rows = rows
+
+    def all(self) -> list[tuple[Any, ...]]:
+        return self._rows
+
+    def scalar_one_or_none(self) -> Any:
+        return self._rows[0][0] if self._rows else None
+
+    def scalars(self) -> '_RowsResult':
+        return self
+
+
+class GraphSession:
+    """仅在内存行集上执行闭包 SELECT，不连接数据库。"""
+
+    def __init__(self, rows: dict[str, dict[int, SimpleNamespace]]) -> None:
+        self.rows = rows
+
+    async def execute(self, statement: Any) -> _RowsResult:
+        columns = list(statement.selected_columns)
+        table_name = columns[0].table.name
+        sql = str(statement.compile(compile_kwargs={'literal_binds': True}))
+        selected = [
+            tuple(getattr(row, column.name) for column in columns)
+            for row in self.rows.get(table_name, {}).values()
+            if self._matches(row, sql)
+        ]
+        return _RowsResult(selected)
+
+    @staticmethod
+    def _matches(row: SimpleNamespace, sql: str) -> bool:
+        for column, values in re.findall(r'\w+\.(\w+) IN \(([^)]+)\)', sql):
+            expected = {value.strip().strip("'") for value in values.split(',')}
+            actual = getattr(row, column)
+            if str(actual) not in expected:
+                return False
+        for column, value in re.findall(r"\w+\.(\w+) = ('[^']*'|\d+)", sql):
+            expected = value.strip("'")
+            if str(getattr(row, column)) != expected:
+                return False
+        return True
+
+
+def _row(**values: Any) -> SimpleNamespace:
+    return SimpleNamespace(**values)
+
+
+def make_batch(**overrides: Any) -> SimpleNamespace:
+    values = {
+        'batch_id': 7,
+        'batch_name': '旧数据回归批次',
+        'status': 'ACTIVE',
+        'preview_action': None,
+        'preview_token_hash': None,
+        'preview_scope_hash': None,
+        'preview_expires_at': None,
+        'update_by': 'admin',
+        'update_time': datetime(2026, 9, 23, 9, 0, 0),
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def preview_command() -> AdoptionPreviewRequestModel:
+    return AdoptionPreviewRequestModel(roots=[AdoptionRootModel(resourceType='pallet', resourceIds=[10])])
+
+
+def execute_command(token: str) -> AdoptionExecuteRequestModel:
+    return AdoptionExecuteRequestModel(
+        batchName='旧数据回归批次',
+        previewToken=token,
+        roots=[AdoptionRootModel(resourceType='pallet', resourceIds=[10])],
+    )
+
+
+class TestBatchGraphTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        self.rows = {table_name: {} for table_name in BATCH_MODELS}
+        self.rows['biz_strain'][1] = _row(strain_id=1, test_batch_id=None, strain_code='S-1')
+        self.rows['biz_warehouse'][2] = _row(warehouse_id=2, test_batch_id=None, warehouse_code='W-2')
+        self.rows['biz_customer'][4] = _row(customer_id=4, test_batch_id=None, customer_code='C-4')
+        self.rows['biz_customer_site'][3] = _row(site_id=3, test_batch_id=None, customer_id=4, site_code='SITE-3')
+        self.rows['biz_pallet'][10] = _row(pallet_id=10, test_batch_id=None, pallet_code='P-10')
+        self.rows['biz_filling_record'][21] = _row(
+            filling_id=21,
+            test_batch_id=None,
+            filling_no='F-21',
+            pallet_id=10,
+            warehouse_id=2,
+            strain_id=1,
+        )
+        self.rows['biz_shipment_order'][30] = _row(
+            shipment_id=30,
+            test_batch_id=None,
+            shipment_no='SO-30',
+            source_warehouse_id=2,
+            target_site_id=3,
+        )
+        self.rows['biz_shipment_item'][31] = _row(
+            shipment_item_id=31,
+            test_batch_id=None,
+            shipment_id=30,
+            pallet_id=10,
+            active_filling_id=21,
+        )
+        self.rows['biz_pallet_inventory'][10] = _row(
+            pallet_id=10,
+            test_batch_id=None,
+            active_filling_id=21,
+            warehouse_id=None,
+            site_id=3,
+            current_line_id=None,
+            last_event_id=41,
+        )
+        self.rows['biz_pallet_event'][41] = _row(
+            event_id=41,
+            test_batch_id=None,
+            pallet_id=10,
+            source_type='SHIPMENT',
+            source_id=30,
+            before_warehouse_id=2,
+            before_site_id=None,
+            before_line_id=None,
+            after_warehouse_id=None,
+            after_site_id=3,
+            after_line_id=None,
+        )
+        self.db = GraphSession(self.rows)
+
+    async def test_closure_walks_foreign_keys_in_both_directions_until_stable(self) -> None:
+        roots = [AdoptionRootModel(resourceType='pallet', resourceIds=[10])]
+
+        closure = await TestBatchGraph.compute_closure(self.db, roots)
+
+        self.assertEqual({10}, closure['biz_pallet'])
+        self.assertEqual({21}, closure['biz_filling_record'])
+        self.assertEqual({31}, closure['biz_shipment_item'])
+        self.assertEqual({30}, closure['biz_shipment_order'])
+        self.assertEqual({41}, closure['biz_pallet_event'])
+        self.assertEqual({3}, closure['biz_customer_site'])
+        self.assertEqual({4}, closure['biz_customer'])
+
+    async def test_closure_deduplicates_cycles(self) -> None:
+        roots = [
+            AdoptionRootModel(resourceType='pallet', resourceIds=[10, 10]),
+            AdoptionRootModel(resourceType='shipment', resourceIds=[30]),
+        ]
+
+        closure = await TestBatchGraph.compute_closure(self.db, roots)
+
+        self.assertEqual({10}, closure['biz_pallet'])
+        self.assertEqual({30}, closure['biz_shipment_order'])
+        self.assertEqual({41}, closure['biz_pallet_event'])
+
+    async def test_event_source_polymorphic_edge_is_walked_in_reverse(self) -> None:
+        roots = [AdoptionRootModel(resourceType='shipment', resourceIds=[30])]
+
+        closure = await TestBatchGraph.compute_closure(self.db, roots)
+
+        self.assertEqual({41}, closure['biz_pallet_event'])
+        self.assertEqual({10}, closure['biz_pallet'])
+
+
+class TestBatchAdoptionContractTest(unittest.TestCase):
+    def test_public_service_signatures_are_stable(self) -> None:
+        expected = {
+            'preview': {
+                'db': AsyncSession,
+                'batch_id': int,
+                'command': AdoptionPreviewRequestModel,
+                'current_user': CurrentUserModel,
+                'return': TestBatchPreviewModel,
+            },
+            'execute': {
+                'db': AsyncSession,
+                'batch_id': int,
+                'command': AdoptionExecuteRequestModel,
+                'current_user': CurrentUserModel,
+                'return': CrudResponseModel,
+            },
+        }
+        for method_name, hints in expected.items():
+            method = getattr(TestBatchAdoptionService, method_name)
+            with self.subTest(method=method_name):
+                self.assertEqual(
+                    ('db', 'batch_id', 'command', 'current_user'),
+                    tuple(inspect.signature(method).parameters),
+                )
+                self.assertEqual(hints, get_type_hints(method))
+
+    def test_selection_summary_is_canonical_and_stable(self) -> None:
+        first = {'biz_pallet': frozenset({10, 2}), 'biz_usage_record': frozenset()}
+        second = {'biz_usage_record': frozenset(), 'biz_pallet': frozenset({2, 10})}
+
+        self.assertEqual(b'{"biz_pallet":[2,10]}', canonical_selection(first))
+        self.assertEqual(selection_hash(first), selection_hash(second))
+
+
+class TestBatchAdoptionServiceTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        self.db = AsyncMock(spec=AsyncSession)
+        self.selection: BatchSelection = {table_name: frozenset() for table_name in BATCH_MODELS}
+        self.selection['biz_pallet'] = frozenset({10})
+        self.selection['biz_usage_record'] = frozenset({50})
+
+    @contextmanager
+    def _preview_stack(
+        self,
+        batch: SimpleNamespace,
+        *,
+        tagged: tuple[str, int] | None = None,
+    ) -> Any:
+        keys = {table_name: [] for table_name in BATCH_MODELS}
+        keys['biz_pallet'] = ['P-10']
+        keys['biz_usage_record'] = ['U-50']
+        contexts = (
+            patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)),
+            patch.object(TestBatchGraph, 'compute_closure', AsyncMock(return_value=self.selection)),
+            patch.object(TestBatchAdoptionDao, 'find_tagged_record', AsyncMock(return_value=tagged)),
+            patch.object(TestBatchAdoptionDao, 'list_display_keys', AsyncMock(return_value=keys)),
+        )
+        with ExitStack() as stack:
+            for context in contexts:
+                stack.enter_context(context)
+            yield
+
+    async def test_preview_persists_ten_minute_single_use_token_and_all_impacts(self) -> None:
+        batch = make_batch()
+        before = datetime.now()
+        with self._preview_stack(batch):
+            result = await TestBatchAdoptionService.preview(self.db, 7, preview_command(), make_admin())
+        after = datetime.now()
+
+        self.assertEqual(7, result.batch_id)
+        self.assertEqual(2, result.total_count)
+        self.assertEqual(list(BATCH_MODELS), [impact.table_name for impact in result.impacts])
+        self.assertEqual(
+            [('P-10',), ('U-50',)],
+            [tuple(impact.keys) for impact in result.impacts if impact.count],
+        )
+        self.assertNotEqual(result.preview_token, batch.preview_token_hash)
+        self.assertEqual('ADOPT', batch.preview_action)
+        self.assertEqual(selection_hash(self.selection), batch.preview_scope_hash)
+        self.assertGreaterEqual(batch.preview_expires_at, before + timedelta(seconds=599))
+        self.assertLessEqual(batch.preview_expires_at, after + timedelta(seconds=601))
+        self.db.commit.assert_awaited_once_with()
+
+    async def test_closure_rejects_any_already_tagged_record(self) -> None:
+        batch = make_batch()
+        with (
+            self._preview_stack(batch, tagged=('biz_usage_record', 50)),
+            self.assertRaisesRegex(ServiceWarning, '已属于其他测试批次'),
+        ):
+            await TestBatchAdoptionService.preview(self.db, 7, preview_command(), make_admin())
+
+        self.db.rollback.assert_awaited_once_with()
+        self.db.commit.assert_not_awaited()
+
+    async def test_regular_user_is_rejected_before_locking_batch(self) -> None:
+        lock_batch = AsyncMock()
+        with (
+            patch.object(TestBatchDao, 'get_batch_for_update', lock_batch),
+            self.assertRaisesRegex(PermissionException, '仅超级管理员'),
+        ):
+            await TestBatchAdoptionService.preview(
+                self.db,
+                7,
+                preview_command(),
+                make_current_user(2, ['system:testbatch:adopt']),
+            )
+
+        lock_batch.assert_not_awaited()
+
+    async def test_execute_recomputes_and_rejects_changed_scope(self) -> None:
+        preview_batch = make_batch()
+        with self._preview_stack(preview_batch):
+            preview = await TestBatchAdoptionService.preview(self.db, 7, preview_command(), make_admin())
+        self.db.reset_mock()
+        changed = dict(self.selection)
+        changed['biz_pallet_event'] = frozenset({41})
+        set_batch = AsyncMock()
+
+        with (
+            patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=preview_batch)),
+            patch.object(TestBatchAdoptionDao, 'lock_roots', AsyncMock()),
+            patch.object(TestBatchAdoptionDao, 'lock_selection', AsyncMock()),
+            patch.object(TestBatchGraph, 'compute_closure', AsyncMock(side_effect=[self.selection, changed])),
+            patch.object(TestBatchAdoptionDao, 'set_test_batch_id', set_batch),
+            self.assertRaisesRegex(ServiceWarning, '预览范围已变化'),
+        ):
+            await TestBatchAdoptionService.execute(self.db, 7, execute_command(preview.preview_token), make_admin())
+
+        set_batch.assert_not_awaited()
+        self.db.rollback.assert_awaited_once_with()
+        self.db.commit.assert_not_awaited()
+
+    async def test_execute_rejects_batch_name_mismatch(self) -> None:
+        batch = make_batch(
+            preview_action='ADOPT',
+            preview_token_hash='unused',
+            preview_scope_hash=selection_hash(self.selection),
+            preview_expires_at=datetime.now() + timedelta(minutes=10),
+        )
+        command = execute_command('x' * 32)
+        command.batch_name = '错误批次名'
+        with (
+            patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)),
+            self.assertRaisesRegex(ServiceWarning, '测试批次名称不匹配'),
+        ):
+            await TestBatchAdoptionService.execute(self.db, 7, command, make_admin())
+
+        self.db.rollback.assert_awaited_once_with()
+
+    async def test_execute_rejects_expired_token(self) -> None:
+        raw_token = 'p' * 43
+        batch = make_batch(
+            preview_action='ADOPT',
+            preview_token_hash=__import__('hashlib').sha256(raw_token.encode()).hexdigest(),
+            preview_scope_hash=selection_hash(self.selection),
+            preview_expires_at=datetime.now() - timedelta(seconds=1),
+        )
+        with (
+            patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)),
+            self.assertRaisesRegex(ServiceWarning, '预览令牌已过期'),
+        ):
+            await TestBatchAdoptionService.execute(self.db, 7, execute_command(raw_token), make_admin())
+
+        self.db.rollback.assert_awaited_once_with()
+
+    async def test_execute_updates_every_nonempty_table_in_stable_order_and_consumes_token(self) -> None:
+        raw_token = 'q' * 43
+        batch = make_batch(
+            preview_action='ADOPT',
+            preview_token_hash=__import__('hashlib').sha256(raw_token.encode()).hexdigest(),
+            preview_scope_hash=selection_hash(self.selection),
+            preview_expires_at=datetime.now() + timedelta(minutes=10),
+        )
+        update = AsyncMock()
+        with (
+            patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)),
+            patch.object(TestBatchAdoptionDao, 'lock_roots', AsyncMock()),
+            patch.object(TestBatchAdoptionDao, 'lock_selection', AsyncMock()),
+            patch.object(TestBatchGraph, 'compute_closure', AsyncMock(return_value=self.selection)),
+            patch.object(TestBatchAdoptionDao, 'find_tagged_record', AsyncMock(return_value=None)),
+            patch.object(TestBatchAdoptionDao, 'set_test_batch_id', update),
+        ):
+            result = await TestBatchAdoptionService.execute(self.db, 7, execute_command(raw_token), make_admin())
+
+        self.assertEqual(CrudResponseModel(is_success=True, message='旧测试数据纳入成功'), result)
+        self.assertEqual(
+            [(self.db, table_name, sorted(self.selection[table_name]), 7) for table_name in sorted(self.selection)],
+            [call.args for call in update.await_args_list],
+        )
+        self.assertIsNone(batch.preview_action)
+        self.assertIsNone(batch.preview_token_hash)
+        self.assertIsNone(batch.preview_scope_hash)
+        self.assertIsNone(batch.preview_expires_at)
+        self.db.commit.assert_awaited_once_with()
+
+        self.db.reset_mock()
+        with (
+            patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)),
+            self.assertRaisesRegex(ServiceWarning, '预览令牌无效'),
+        ):
+            await TestBatchAdoptionService.execute(self.db, 7, execute_command(raw_token), make_admin())
+        self.db.rollback.assert_awaited_once_with()
+
+    async def test_any_update_failure_rolls_back_entire_adoption(self) -> None:
+        raw_token = 'r' * 43
+        batch = make_batch(
+            preview_action='ADOPT',
+            preview_token_hash=__import__('hashlib').sha256(raw_token.encode()).hexdigest(),
+            preview_scope_hash=selection_hash(self.selection),
+            preview_expires_at=datetime.now() + timedelta(minutes=10),
+        )
+        update = AsyncMock(side_effect=[None, RuntimeError('update failed')])
+        with (
+            patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)),
+            patch.object(TestBatchAdoptionDao, 'lock_roots', AsyncMock()),
+            patch.object(TestBatchAdoptionDao, 'lock_selection', AsyncMock()),
+            patch.object(TestBatchGraph, 'compute_closure', AsyncMock(return_value=self.selection)),
+            patch.object(TestBatchAdoptionDao, 'find_tagged_record', AsyncMock(return_value=None)),
+            patch.object(TestBatchAdoptionDao, 'set_test_batch_id', update),
+            self.assertRaisesRegex(RuntimeError, 'update failed'),
+        ):
+            await TestBatchAdoptionService.execute(self.db, 7, execute_command(raw_token), make_admin())
+
+        self.db.rollback.assert_awaited_once_with()
+        self.db.commit.assert_not_awaited()
+
+
+if __name__ == '__main__':
+    unittest.main()