Files

549 lines
22 KiB
Python

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_tunnel': '隧道',
'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
self.statements: list[Any] = []
async def execute(self, statement: Any) -> _RowsResult:
self.statements.append(statement)
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)
class CommitExpiringBatch:
"""模拟 AsyncSession commit 后 ORM 属性过期。"""
def __init__(self) -> None:
self.expired = False
self._batch_name = '旧数据回归批次'
self.batch_id = 7
self.status = 'ACTIVE'
self.preview_action = None
self.preview_token_hash = None
self.preview_scope_hash = None
self.preview_expires_at = None
self.update_by = 'admin'
self.update_time = datetime(2026, 9, 23, 9, 0, 0)
@property
def batch_name(self) -> str:
if self.expired:
raise RuntimeError('commit 后不得读取过期 ORM 属性')
return self._batch_name
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,
tunnel_id=None,
trace_no=None,
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_tunnel_id=None,
before_line_id=None,
after_warehouse_id=None,
after_site_id=3,
after_tunnel_id=None,
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'])
async def test_execute_closure_uses_for_update_current_reads(self) -> None:
roots = [AdoptionRootModel(resourceType='pallet', resourceIds=[10])]
await TestBatchGraph.compute_locked_closure(self.db, roots)
self.assertTrue(self.db.statements)
for statement in self.db.statements:
with self.subTest(sql=str(statement)):
self.assertIn('FOR UPDATE', str(statement.compile()).upper())
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(
list(TABLE_LABELS.items()),
[(impact.table_name, impact.label) 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_preview_materializes_response_before_commit_expires_batch(self) -> None:
batch = CommitExpiringBatch()
async def expire_batch() -> None:
batch.expired = True
self.db.commit.side_effect = expire_batch
with self._preview_stack(batch):
result = await TestBatchAdoptionService.preview(self.db, 7, preview_command(), make_admin())
self.assertEqual('旧数据回归批次', result.batch_name)
self.db.commit.assert_awaited_once_with()
self.db.rollback.assert_not_awaited()
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_preview_rejects_when_any_requested_root_does_not_exist(self) -> None:
batch = make_batch()
command = AdoptionPreviewRequestModel(roots=[AdoptionRootModel(resourceType='pallet', resourceIds=[10, 999])])
with (
self._preview_stack(batch),
self.assertRaisesRegex(ServiceWarning, '指定的旧数据不存在'),
):
await TestBatchAdoptionService.preview(self.db, 7, command, make_admin())
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.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_locked_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_closed_batch_before_business_row_access(self) -> None:
lock_roots = AsyncMock()
compute_closure = AsyncMock()
lock_selection = AsyncMock()
update = AsyncMock()
with (
patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=make_batch(status='CLOSED'))),
patch.object(TestBatchAdoptionDao, 'lock_roots', lock_roots),
patch.object(TestBatchGraph, 'compute_locked_closure', compute_closure),
patch.object(TestBatchAdoptionDao, 'lock_selection', lock_selection),
patch.object(TestBatchAdoptionDao, 'set_test_batch_id', update),
self.assertRaisesRegex(ServiceWarning, '测试批次已关闭'),
):
await TestBatchAdoptionService.execute(
self.db,
7,
execute_command('x' * 32),
make_admin(),
)
lock_roots.assert_not_awaited()
compute_closure.assert_not_awaited()
lock_selection.assert_not_awaited()
update.assert_not_awaited()
self.db.rollback.assert_awaited_once_with()
self.db.commit.assert_not_awaited()
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_locked_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_locked_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()