diff --git a/shierfeng-fastapi-backend/module_business/dao/test_batch_adoption_dao.py b/shierfeng-fastapi-backend/module_business/dao/test_batch_adoption_dao.py new file mode 100644 index 0000000..e2f2958 --- /dev/null +++ b/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('', '存在记录已属于其他测试批次') diff --git a/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py b/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py index 49b2a31..c95c48c 100644 --- a/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py +++ b/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( - select(func.count()).select_from(model).where(model.test_batch_id == batch_id) - ) + result = await db.execute(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 diff --git a/shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py b/shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py new file mode 100644 index 0000000..1725c88 --- /dev/null +++ b/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} diff --git a/shierfeng-fastapi-backend/module_business/service/test_batch_adoption_service.py b/shierfeng-fastapi-backend/module_business/service/test_batch_adoption_service.py new file mode 100644 index 0000000..5301828 --- /dev/null +++ b/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 diff --git a/shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py b/shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py new file mode 100644 index 0000000..69c1d34 --- /dev/null +++ b/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()