544 lines
22 KiB
Python
544 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_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,
|
|
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'])
|
|
|
|
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()
|