diff --git a/shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py b/shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py index 1725c88..b72ce0d 100644 --- a/shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py +++ b/shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py @@ -214,18 +214,29 @@ class TestBatchGraph: return edges @staticmethod - async def _existing_ids(db: AsyncSession, table_name: str, ids: set[int]) -> set[int]: + async def _existing_ids( + db: AsyncSession, + table_name: str, + ids: set[int], + *, + for_update: bool, + ) -> 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)) + statement = select(primary_key).where(primary_key.in_(sorted(ids))).order_by(primary_key) + if for_update: + statement = statement.with_for_update() + result = await db.execute(statement) return {row[0] for row in result.all()} @classmethod - async def compute_closure( + async def _compute_closure( cls, db: AsyncSession, roots: list[AdoptionRootModel], + *, + for_update: bool, ) -> BatchSelection: selected: dict[str, set[int]] = {table_name: set() for table_name in BATCH_MODELS} requested: dict[str, set[int]] = defaultdict(set) @@ -234,7 +245,12 @@ class TestBatchGraph: queue: deque[str] = deque() for table_name in sorted(requested): - existing = await cls._existing_ids(db, table_name, requested[table_name]) + existing = await cls._existing_ids( + db, + table_name, + requested[table_name], + for_update=for_update, + ) selected[table_name].update(existing) if existing: queue.append(table_name) @@ -251,7 +267,10 @@ class TestBatchGraph: 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)) + statement = statement.order_by(edge.result_column) + if for_update: + statement = statement.with_for_update() + result = await db.execute(statement) 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: @@ -259,3 +278,20 @@ class TestBatchGraph: queue.append(edge.target_table) return {table_name: frozenset(selected[table_name]) for table_name in BATCH_MODELS} + + @classmethod + async def compute_closure( + cls, + db: AsyncSession, + roots: list[AdoptionRootModel], + ) -> BatchSelection: + return await cls._compute_closure(db, roots, for_update=False) + + @classmethod + async def compute_locked_closure( + cls, + db: AsyncSession, + roots: list[AdoptionRootModel], + ) -> BatchSelection: + """使用锁定读计算执行阶段闭包，避免复用一致性快照。""" + return await cls._compute_closure(db, roots, for_update=True) 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 index 5301828..0a2a027 100644 --- a/shierfeng-fastapi-backend/module_business/service/test_batch_adoption_service.py +++ b/shierfeng-fastapi-backend/module_business/service/test_batch_adoption_service.py @@ -97,7 +97,6 @@ class TestBatchAdoptionService: update_by=user_name, update_time=now, ) - await db.commit() impacts = [ BatchImpactModel( table_name=table_name, @@ -107,7 +106,7 @@ class TestBatchAdoptionService: ) for table_name in BATCH_MODELS ] - return TestBatchPreviewModel( + response = TestBatchPreviewModel( batch_id=batch_id, batch_name=batch.batch_name, preview_token=raw_token, @@ -115,6 +114,8 @@ class TestBatchAdoptionService: total_count=sum(item.count for item in impacts), impacts=impacts, ) + await db.commit() + return response except Exception: await db.rollback() raise @@ -137,9 +138,9 @@ class TestBatchAdoptionService: 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) + first_selection = await TestBatchGraph.compute_locked_closure(db, command.roots) await TestBatchAdoptionDao.lock_selection(db, first_selection) - locked_selection = await TestBatchGraph.compute_closure(db, command.roots) + locked_selection = await TestBatchGraph.compute_locked_closure(db, command.roots) expected_scope_hash = batch.preview_scope_hash if ( selection_hash(first_selection) != expected_scope_hash 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 index 69c1d34..384e981 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_test_batch_adoption.py @@ -69,8 +69,10 @@ class GraphSession: 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})) @@ -115,6 +117,28 @@ def make_batch(**overrides: Any) -> SimpleNamespace: 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])]) @@ -214,6 +238,16 @@ class TestBatchGraphTest(unittest.IsolatedAsyncioTestCase): 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: @@ -288,6 +322,10 @@ class TestBatchAdoptionServiceTest(unittest.IsolatedAsyncioTestCase): 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], @@ -299,6 +337,20 @@ class TestBatchAdoptionServiceTest(unittest.IsolatedAsyncioTestCase): 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 ( @@ -338,7 +390,7 @@ class TestBatchAdoptionServiceTest(unittest.IsolatedAsyncioTestCase): 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(TestBatchGraph, 'compute_locked_closure', AsyncMock(side_effect=[self.selection, changed])), patch.object(TestBatchAdoptionDao, 'set_test_batch_id', set_batch), self.assertRaisesRegex(ServiceWarning, '预览范围已变化'), ): @@ -365,6 +417,33 @@ class TestBatchAdoptionServiceTest(unittest.IsolatedAsyncioTestCase): 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( @@ -394,7 +473,7 @@ class TestBatchAdoptionServiceTest(unittest.IsolatedAsyncioTestCase): 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(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), ): @@ -432,7 +511,7 @@ class TestBatchAdoptionServiceTest(unittest.IsolatedAsyncioTestCase): 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(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'),