From af5428d0e384683a918a5819b41b0cb2b1f05633 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Tue, 22 Sep 2026 16:19:32 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E6=89=B9=E6=AC=A1=E7=94=9F=E5=91=BD=E5=91=A8?= =?UTF-8?q?=E6=9C=9F=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/dao/test_batch_dao.py | 111 ++- .../entity/vo/test_batch_vo.py | 3 +- .../service/test_batch_service.py | 322 +++++++++ .../test_test_batch_service.py | 661 ++++++++++++++++++ 4 files changed, 1095 insertions(+), 2 deletions(-) create mode 100644 shierfeng-fastapi-backend/module_business/service/test_batch_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_test_batch_service.py 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 3f0e0d2..49b2a31 100644 --- a/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py @@ -1,9 +1,52 @@ -from sqlalchemy import select +from datetime import datetime +from typing import Any + +from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession +from module_admin.entity.do.user_do import SysUser from module_business.domain.enums import TestBatchStatus +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.do.test_batch_do import BizTestBatch, BizTestBatchUser +_BATCH_OWNED_MODELS: tuple[type[Any], ...] = ( + BizStrain, + BizCustomer, + BizCustomerSite, + BizProductionLine, + BizWarehouse, + BizPallet, + BizInboundOrder, + BizInboundItem, + BizFillingRecord, + BizShipmentOrder, + BizShipmentItem, + BizUsageRecord, + BizReturnOrder, + BizReturnItem, + BizPalletInventory, + BizPalletEvent, +) + class TestBatchDao: """测试批次访问数据库操作层。""" @@ -45,3 +88,69 @@ class TestBatchDao: statement = statement.join(BizTestBatchUser).where(BizTestBatchUser.user_id == user_id) result = await db.execute(statement.order_by(BizTestBatch.batch_id.desc())) return list(result.scalars().all()) + + @classmethod + async def list_batches(cls, db: AsyncSession) -> list[BizTestBatch]: + result = await db.execute(select(BizTestBatch).order_by(BizTestBatch.batch_id.desc())) + return list(result.scalars().all()) + + @classmethod + async def add_batch(cls, db: AsyncSession, batch: BizTestBatch) -> None: + db.add(batch) + + @classmethod + async def list_batch_user_ids(cls, db: AsyncSession, batch_id: int) -> list[int]: + result = await db.execute( + select(BizTestBatchUser.user_id) + .where(BizTestBatchUser.batch_id == batch_id) + .order_by(BizTestBatchUser.user_id) + ) + return list(result.scalars().all()) + + @classmethod + async def list_active_user_ids(cls, db: AsyncSession, user_ids: list[int]) -> list[int]: + if not user_ids: + return [] + result = await db.execute( + select(SysUser.user_id) + .where( + SysUser.user_id.in_(user_ids), + SysUser.status == '0', + SysUser.del_flag == '0', + ) + .order_by(SysUser.user_id) + ) + return list(result.scalars().all()) + + @classmethod + async def replace_batch_users( + cls, + db: AsyncSession, + batch_id: int, + user_ids: list[int], + create_by: str, + create_time: datetime, + ) -> None: + await db.execute(delete(BizTestBatchUser).where(BizTestBatchUser.batch_id == batch_id)) + if user_ids: + db.add_all( + [ + BizTestBatchUser( + batch_id=batch_id, + user_id=user_id, + create_by=create_by, + create_time=create_time, + ) + for user_id in user_ids + ] + ) + + @classmethod + 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) + ) + counts[model.__tablename__] = result.scalar() or 0 + return counts diff --git a/shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py b/shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py index 1595907..c4cf773 100644 --- a/shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py +++ b/shierfeng-fastapi-backend/module_business/entity/vo/test_batch_vo.py @@ -7,6 +7,7 @@ from pydantic.alias_generators import to_camel from module_business.domain.enums import TestBatchStatus PositiveId = Annotated[int, Field(gt=0)] +NonNegativeCount = Annotated[int, Field(ge=0)] class TestBatchApiModel(BaseModel): @@ -45,7 +46,7 @@ class TestBatchModel(TestBatchApiModel): closed_at: datetime | None = None purged_by: int | None = None purged_at: datetime | None = None - purge_summary: dict[str, int] | None = None + purge_summary: dict[str, NonNegativeCount] | None = None class TestBatchOptionModel(TestBatchApiModel): diff --git a/shierfeng-fastapi-backend/module_business/service/test_batch_service.py b/shierfeng-fastapi-backend/module_business/service/test_batch_service.py new file mode 100644 index 0000000..2574874 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/test_batch_service.py @@ -0,0 +1,322 @@ +import json +import secrets +from datetime import datetime +from typing import Any + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from common.vo import CrudResponseModel +from exceptions.exception import PermissionException, ServiceException, ServiceWarning +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.dao.test_batch_dao import TestBatchDao +from module_business.domain.enums import TestBatchStatus +from module_business.entity.do.test_batch_do import BizTestBatch +from module_business.entity.vo.test_batch_vo import ( + BatchImpactModel, + TestBatchCreateModel, + TestBatchDetailModel, + TestBatchEditModel, + TestBatchModel, + TestBatchOptionModel, + TestBatchUserAssignModel, +) +from module_business.service.test_batch_access_service import TestBatchAccessService + +_OWNED_TABLE_LABELS: tuple[tuple[str, str], ...] = ( + ('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 TestBatchService: + """测试数据批次生命周期与用户授权服务。""" + + @staticmethod + def _assert_admin(current_user: CurrentUserModel) -> None: + TestBatchAccessService.assert_super_admin(current_user) + + @staticmethod + def _user_identity(current_user: CurrentUserModel) -> tuple[int, str]: + user = current_user.user + if user is None or user.user_id is None or user.user_name is None: + raise PermissionException('', '当前用户信息不可用') + return user.user_id, user.user_name + + @staticmethod + def _parse_purge_summary(raw_summary: Any) -> dict[str, int] | None: + if raw_summary is None: + return None + try: + parsed = json.loads(raw_summary) if isinstance(raw_summary, str) else raw_summary + except (TypeError, ValueError): + raise ServiceException(message='测试批次清理摘要格式无效') from None + if not isinstance(parsed, dict) or any( + not isinstance(key, str) + or isinstance(value, bool) + or not isinstance(value, int) + or value < 0 + for key, value in parsed.items() + ): + raise ServiceException(message='测试批次清理摘要格式无效') + return parsed + + @classmethod + def _to_batch_model(cls, batch: BizTestBatch | Any) -> TestBatchModel: + values = { + field_name: getattr(batch, field_name, None) + for field_name in TestBatchModel.model_fields + } + values['purge_summary'] = cls._parse_purge_summary(getattr(batch, 'purge_summary', None)) + return TestBatchModel.model_validate(values) + + @staticmethod + def _clear_preview(batch: BizTestBatch | Any) -> None: + batch.preview_action = None + batch.preview_token_hash = None + batch.preview_scope_hash = None + batch.preview_expires_at = None + + @staticmethod + def _touch(batch: BizTestBatch | Any, user_name: str, now: datetime) -> None: + batch.update_by = user_name + batch.update_time = now + + @staticmethod + def _raise_for_non_active(batch: BizTestBatch | Any) -> None: + if batch.status == TestBatchStatus.CLOSED.value: + raise ServiceWarning(message='测试批次已关闭') + if batch.status == TestBatchStatus.PURGED.value: + raise ServiceWarning(message='测试批次已清理') + if batch.status != TestBatchStatus.ACTIVE.value: + raise ServiceWarning(message='测试批次状态异常') + + @classmethod + async def list_available_batches( + cls, + db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[TestBatchOptionModel]: + user_id, _ = cls._user_identity(current_user) + is_admin = current_user.user is not None and current_user.user.admin is True + batches = await TestBatchDao.list_available_batches(db, user_id, is_admin) + return [TestBatchOptionModel.model_validate(batch) for batch in batches] + + @classmethod + async def list_batches( + cls, + db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[TestBatchModel]: + cls._assert_admin(current_user) + batches = await TestBatchDao.list_batches(db) + return [cls._to_batch_model(batch) for batch in batches] + + @classmethod + async def get_batch_detail( + cls, + db: AsyncSession, + batch_id: int, + current_user: CurrentUserModel, + ) -> TestBatchDetailModel: + cls._assert_admin(current_user) + batch = await TestBatchDao.get_batch(db, batch_id) + if batch is None: + raise ServiceWarning(message='测试批次不存在') + + user_ids = await TestBatchDao.list_batch_user_ids(db, batch_id) + counts = await TestBatchDao.count_owned_rows(db, batch_id) + data_summary = [ + BatchImpactModel( + table_name=table_name, + label=label, + count=counts.get(table_name, 0), + keys=[], + ) + for table_name, label in _OWNED_TABLE_LABELS + ] + batch_model = cls._to_batch_model(batch) + return TestBatchDetailModel( + **batch_model.model_dump(), + user_ids=user_ids, + data_summary=data_summary, + ) + + @classmethod + async def create_batch( + cls, + db: AsyncSession, + command: TestBatchCreateModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + cls._assert_admin(current_user) + _, user_name = cls._user_identity(current_user) + batch_name = command.batch_name.strip() + if not batch_name: + raise ServiceWarning(message='测试批次名称不能为空') + + now = datetime.now() + batch = BizTestBatch( + batch_code=f'TB-{now:%Y%m%d%H%M%S}-{secrets.token_hex(4).upper()}', + batch_name=batch_name, + status=TestBatchStatus.ACTIVE.value, + create_by=user_name, + create_time=now, + update_by=user_name, + update_time=now, + remark=command.remark, + ) + try: + await TestBatchDao.add_batch(db, batch) + await db.commit() + return CrudResponseModel(is_success=True, message='测试批次创建成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='测试批次创建失败，批次编码冲突') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def edit_batch( + cls, + db: AsyncSession, + batch_id: int, + command: TestBatchEditModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + cls._assert_admin(current_user) + _, user_name = cls._user_identity(current_user) + try: + batch = await TestBatchDao.get_batch_for_update(db, batch_id) + if batch is None: + raise ServiceWarning(message='测试批次不存在') + if batch.status == TestBatchStatus.PURGED.value: + raise ServiceWarning(message='测试批次已清理，不能修改') + if batch.status not in (TestBatchStatus.ACTIVE.value, TestBatchStatus.CLOSED.value): + raise ServiceWarning(message='测试批次状态异常') + + now = datetime.now() + batch.remark = command.remark + cls._touch(batch, user_name, now) + await db.commit() + return CrudResponseModel(is_success=True, message='测试批次修改成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='测试批次修改失败，数据状态已变化') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def replace_users( + cls, + db: AsyncSession, + batch_id: int, + command: TestBatchUserAssignModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + cls._assert_admin(current_user) + _, user_name = cls._user_identity(current_user) + try: + batch = await TestBatchDao.get_batch_for_update(db, batch_id) + if batch is None: + raise ServiceWarning(message='测试批次不存在') + cls._raise_for_non_active(batch) + + user_ids = list(dict.fromkeys(command.user_ids)) + if user_ids: + active_user_ids = await TestBatchDao.list_active_user_ids(db, user_ids) + if set(active_user_ids) != set(user_ids): + raise ServiceWarning(message='存在不存在、停用或已删除的用户') + + now = datetime.now() + await TestBatchDao.replace_batch_users(db, batch_id, user_ids, user_name, now) + cls._touch(batch, user_name, now) + await db.commit() + return CrudResponseModel(is_success=True, message='测试批次用户授权成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='测试批次用户授权失败，关联数据无效') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def close_batch( + cls, + db: AsyncSession, + batch_id: int, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + cls._assert_admin(current_user) + user_id, user_name = cls._user_identity(current_user) + try: + batch = await TestBatchDao.get_batch_for_update(db, batch_id) + if batch is None: + raise ServiceWarning(message='测试批次不存在') + cls._raise_for_non_active(batch) + + now = datetime.now() + batch.status = TestBatchStatus.CLOSED.value + batch.closed_by = user_id + batch.closed_at = now + cls._clear_preview(batch) + cls._touch(batch, user_name, now) + await db.commit() + return CrudResponseModel(is_success=True, message='测试批次关闭成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='测试批次关闭失败，数据状态已变化') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def reopen_batch( + cls, + db: AsyncSession, + batch_id: int, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + cls._assert_admin(current_user) + _, user_name = cls._user_identity(current_user) + try: + batch = await TestBatchDao.get_batch_for_update(db, batch_id) + if batch is None: + raise ServiceWarning(message='测试批次不存在') + if batch.status == TestBatchStatus.PURGED.value: + raise ServiceWarning(message='测试批次已清理，不能重新开启') + if batch.status == TestBatchStatus.ACTIVE.value: + raise ServiceWarning(message='测试批次已处于启用状态') + if batch.status != TestBatchStatus.CLOSED.value: + raise ServiceWarning(message='测试批次状态异常') + + now = datetime.now() + batch.status = TestBatchStatus.ACTIVE.value + batch.closed_by = None + batch.closed_at = None + cls._clear_preview(batch) + cls._touch(batch, user_name, now) + await db.commit() + return CrudResponseModel(is_success=True, message='测试批次重新开启成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='测试批次重新开启失败，数据状态已变化') from None + except Exception: + await db.rollback() + raise diff --git a/shierfeng-fastapi-backend/tests/module_business/test_test_batch_service.py b/shierfeng-fastapi-backend/tests/module_business/test_test_batch_service.py new file mode 100644 index 0000000..863ec23 --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_test_batch_service.py @@ -0,0 +1,661 @@ +import inspect +import json +import unittest +from datetime import datetime +from types import SimpleNamespace +from typing import get_type_hints +from unittest.mock import AsyncMock, MagicMock, patch + +from sqlalchemy.dialects import postgresql +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from common.vo import CrudResponseModel +from exceptions.exception import PermissionException, ServiceException, ServiceWarning +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.dao.test_batch_dao import TestBatchDao +from module_business.entity.do.test_batch_do import BizTestBatch, BizTestBatchUser +from module_business.entity.vo.test_batch_vo import ( + TestBatchCreateModel, + TestBatchDetailModel, + TestBatchEditModel, + TestBatchModel, + TestBatchOptionModel, + TestBatchUserAssignModel, +) +from module_business.service.test_batch_service import TestBatchService +from tests.module_business.test_batch_fixtures import make_admin, make_current_user + +OWNED_TABLES = ( + ('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', '托盘事件'), +) + + +def make_batch( + *, + batch_id: int = 3, + status: str = 'ACTIVE', + purge_summary: str | None = None, +) -> SimpleNamespace: + created_at = datetime(2026, 9, 22, 9, 0, 0) + return SimpleNamespace( + batch_id=batch_id, + batch_code=f'TB-{batch_id}', + batch_name=f'测试批次-{batch_id}', + status=status, + remark='回归测试', + create_by='admin', + create_time=created_at, + update_by='admin', + update_time=created_at, + closed_by=None, + closed_at=None, + purged_by=None, + purged_at=None, + purge_summary=purge_summary, + preview_action='ADOPT', + preview_token_hash='SENSITIVE_TOKEN_HASH', + preview_scope_hash='SENSITIVE_SCOPE_HASH', + preview_expires_at=datetime(2026, 9, 22, 10, 0, 0), + ) + + +def compile_sql(statement: object) -> str: + return ' '.join( + str(statement.compile(dialect=postgresql.dialect(), compile_kwargs={'literal_binds': True})).split() + ) + + +class TestBatchServiceSignatureTest(unittest.TestCase): + def test_eight_public_service_signatures_are_stable(self) -> None: + expected = { + 'list_available_batches': ( + ('db', 'current_user'), + {'db': AsyncSession, 'current_user': CurrentUserModel, 'return': list[TestBatchOptionModel]}, + ), + 'list_batches': ( + ('db', 'current_user'), + {'db': AsyncSession, 'current_user': CurrentUserModel, 'return': list[TestBatchModel]}, + ), + 'get_batch_detail': ( + ('db', 'batch_id', 'current_user'), + { + 'db': AsyncSession, + 'batch_id': int, + 'current_user': CurrentUserModel, + 'return': TestBatchDetailModel, + }, + ), + 'create_batch': ( + ('db', 'command', 'current_user'), + { + 'db': AsyncSession, + 'command': TestBatchCreateModel, + 'current_user': CurrentUserModel, + 'return': CrudResponseModel, + }, + ), + 'edit_batch': ( + ('db', 'batch_id', 'command', 'current_user'), + { + 'db': AsyncSession, + 'batch_id': int, + 'command': TestBatchEditModel, + 'current_user': CurrentUserModel, + 'return': CrudResponseModel, + }, + ), + 'replace_users': ( + ('db', 'batch_id', 'command', 'current_user'), + { + 'db': AsyncSession, + 'batch_id': int, + 'command': TestBatchUserAssignModel, + 'current_user': CurrentUserModel, + 'return': CrudResponseModel, + }, + ), + 'close_batch': ( + ('db', 'batch_id', 'current_user'), + { + 'db': AsyncSession, + 'batch_id': int, + 'current_user': CurrentUserModel, + 'return': CrudResponseModel, + }, + ), + 'reopen_batch': ( + ('db', 'batch_id', 'current_user'), + { + 'db': AsyncSession, + 'batch_id': int, + 'current_user': CurrentUserModel, + 'return': CrudResponseModel, + }, + ), + } + + self.assertEqual(set(expected), {name for name in expected if hasattr(TestBatchService, name)}) + for name, (parameters, hints) in expected.items(): + method = getattr(TestBatchService, name) + with self.subTest(method=name): + self.assertTrue(inspect.iscoroutinefunction(method)) + self.assertEqual(parameters, tuple(inspect.signature(method).parameters)) + self.assertEqual(hints, get_type_hints(method)) + + +class TestBatchServiceReadTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.db = AsyncMock(spec=AsyncSession) + + async def test_regular_user_options_only_include_own_active_batches(self) -> None: + batches = [make_batch()] + with patch.object(TestBatchDao, 'list_available_batches', AsyncMock(return_value=batches)) as query: + result = await TestBatchService.list_available_batches(self.db, make_current_user(8)) + + query.assert_awaited_once_with(self.db, 8, False) + self.assertEqual([3], [item.batch_id for item in result]) + self.assertTrue(all(isinstance(item, TestBatchOptionModel) for item in result)) + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_admin_options_request_all_active_batches(self) -> None: + with patch.object(TestBatchDao, 'list_available_batches', AsyncMock(return_value=[])) as query: + result = await TestBatchService.list_available_batches(self.db, make_admin()) + + self.assertEqual([], result) + query.assert_awaited_once_with(self.db, 1, True) + + async def test_options_reject_missing_user_context_without_querying(self) -> None: + current_user = CurrentUserModel(permissions=[], roles=[], user=None) + query = AsyncMock() + with patch.object(TestBatchDao, 'list_available_batches', query), self.assertRaises(PermissionException): + await TestBatchService.list_available_batches(self.db, current_user) + + query.assert_not_awaited() + + async def test_management_list_requires_hard_admin_flag(self) -> None: + query = AsyncMock() + with ( + patch.object(TestBatchDao, 'list_batches', query), + self.assertRaises(PermissionException) as raised, + ): + await TestBatchService.list_batches( + self.db, + make_current_user(2, ['system:testbatch:list']), + ) + + self.assertIn('仅超级管理员', raised.exception.message) + query.assert_not_awaited() + + async def test_list_parses_purge_summary_and_excludes_preview_secrets(self) -> None: + batch = make_batch(status='PURGED', purge_summary='{"biz_pallet": 2, "biz_usage_record": 1}') + with patch.object(TestBatchDao, 'list_batches', AsyncMock(return_value=[batch])): + result = await TestBatchService.list_batches(self.db, make_admin()) + + self.assertEqual({'biz_pallet': 2, 'biz_usage_record': 1}, result[0].purge_summary) + serialized = json.dumps(result[0].model_dump(mode='json', by_alias=True), ensure_ascii=False) + self.assertNotIn('SENSITIVE_TOKEN_HASH', serialized) + self.assertNotIn('SENSITIVE_SCOPE_HASH', serialized) + self.assertNotIn('previewToken', serialized) + + async def test_invalid_purge_summary_is_rejected_without_echoing_content(self) -> None: + sensitive_value = 'RAW_PREVIEW_TOKEN_MUST_NOT_LEAK' + batch = make_batch(status='PURGED', purge_summary=f'{{"token": "{sensitive_value}"}}') + with ( + patch.object(TestBatchDao, 'list_batches', AsyncMock(return_value=[batch])), + self.assertRaises(ServiceException) as raised, + ): + await TestBatchService.list_batches(self.db, make_admin()) + + self.assertEqual('测试批次清理摘要格式无效', raised.exception.message) + self.assertNotIn(sensitive_value, raised.exception.message) + + async def test_missing_batch_detail_stops_before_related_queries(self) -> None: + users = AsyncMock() + counts = AsyncMock() + with ( + patch.object(TestBatchDao, 'get_batch', AsyncMock(return_value=None)), + patch.object(TestBatchDao, 'list_batch_user_ids', users), + patch.object(TestBatchDao, 'count_owned_rows', counts), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.get_batch_detail(self.db, 404, make_admin()) + + self.assertIn('不存在', raised.exception.message) + users.assert_not_awaited() + counts.assert_not_awaited() + + async def test_detail_returns_all_sixteen_classifications_and_user_ids(self) -> None: + counts = {table_name: index for index, (table_name, _) in enumerate(OWNED_TABLES)} + with ( + patch.object(TestBatchDao, 'get_batch', AsyncMock(return_value=make_batch())), + patch.object(TestBatchDao, 'list_batch_user_ids', AsyncMock(return_value=[7, 9])), + patch.object(TestBatchDao, 'count_owned_rows', AsyncMock(return_value=counts)), + ): + result = await TestBatchService.get_batch_detail(self.db, 3, make_admin()) + + self.assertEqual([7, 9], result.user_ids) + self.assertEqual( + [(table_name, label, index, []) for index, (table_name, label) in enumerate(OWNED_TABLES)], + [(item.table_name, item.label, item.count, item.keys) for item in result.data_summary], + ) + serialized = json.dumps(result.model_dump(mode='json', by_alias=True), ensure_ascii=False) + self.assertNotIn('SENSITIVE_TOKEN_HASH', serialized) + + +class TestBatchServiceWriteTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.db = AsyncMock(spec=AsyncSession) + self.admin = make_admin() + + async def test_non_admin_cannot_create_even_with_permission_character(self) -> None: + add_batch = AsyncMock() + command = TestBatchCreateModel(batchName='测试批次') + with patch.object(TestBatchDao, 'add_batch', add_batch), self.assertRaises(PermissionException): + await TestBatchService.create_batch( + self.db, + command, + make_current_user(2, ['system:testbatch:add']), + ) + + add_batch.assert_not_awaited() + self.db.commit.assert_not_awaited() + + async def test_create_trims_name_generates_code_and_commits(self) -> None: + command = TestBatchCreateModel(batchName=' 回归批次 ', remark='用于回归') + with ( + patch('module_business.service.test_batch_service.secrets.token_hex', return_value='a1b2c3d4'), + patch.object(TestBatchDao, 'add_batch', AsyncMock()) as add_batch, + ): + result = await TestBatchService.create_batch(self.db, command, self.admin) + + created = add_batch.await_args.args[1] + self.assertIsInstance(created, BizTestBatch) + self.assertEqual('回归批次', created.batch_name) + self.assertRegex(created.batch_code, r'^TB-\d{14}-A1B2C3D4$') + self.assertEqual('ACTIVE', created.status) + self.assertEqual('admin', created.create_by) + self.assertEqual(CrudResponseModel(is_success=True, message='测试批次创建成功'), result) + self.db.commit.assert_awaited_once_with() + self.db.rollback.assert_not_awaited() + + async def test_blank_trimmed_name_is_rejected_before_write(self) -> None: + add_batch = AsyncMock() + with ( + patch.object(TestBatchDao, 'add_batch', add_batch), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.create_batch( + self.db, + TestBatchCreateModel(batchName=' '), + self.admin, + ) + + self.assertIn('名称不能为空', raised.exception.message) + add_batch.assert_not_awaited() + + async def test_create_integrity_error_rolls_back_and_maps_message(self) -> None: + error = IntegrityError('insert', {}, RuntimeError('duplicate')) + with ( + patch.object(TestBatchDao, 'add_batch', AsyncMock(side_effect=error)), + self.assertRaises(ServiceException) as raised, + ): + await TestBatchService.create_batch( + self.db, + TestBatchCreateModel(batchName='测试批次'), + self.admin, + ) + + self.assertIn('批次编码冲突', raised.exception.message) + self.db.rollback.assert_awaited_once_with() + self.db.commit.assert_not_awaited() + + async def test_unexpected_create_failure_rolls_back_and_propagates(self) -> None: + with ( + patch.object(TestBatchDao, 'add_batch', AsyncMock(side_effect=RuntimeError('database unavailable'))), + self.assertRaisesRegex(RuntimeError, 'database unavailable'), + ): + await TestBatchService.create_batch( + self.db, + TestBatchCreateModel(batchName='测试批次'), + self.admin, + ) + + self.db.rollback.assert_awaited_once_with() + + async def test_edit_missing_batch_rolls_back(self) -> None: + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=None)), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.edit_batch( + self.db, + 3, + TestBatchEditModel(remark='备注'), + self.admin, + ) + + self.assertIn('不存在', raised.exception.message) + self.db.rollback.assert_awaited_once_with() + self.db.commit.assert_not_awaited() + + async def test_edit_purged_batch_is_rejected(self) -> None: + batch = make_batch(status='PURGED') + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.edit_batch( + self.db, + 3, + TestBatchEditModel(remark='备注'), + self.admin, + ) + + self.assertIn('已清理', raised.exception.message) + self.db.rollback.assert_awaited_once_with() + + async def test_edit_active_batch_updates_only_remark_and_audit_fields(self) -> None: + batch = make_batch() + original_name = batch.batch_name + with patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)): + result = await TestBatchService.edit_batch( + self.db, + 3, + TestBatchEditModel(remark='新备注'), + self.admin, + ) + + self.assertEqual('新备注', batch.remark) + self.assertEqual(original_name, batch.batch_name) + self.assertEqual('admin', batch.update_by) + self.assertEqual('测试批次修改成功', result.message) + self.db.commit.assert_awaited_once_with() + + async def test_replace_users_locks_first_deduplicates_and_commits(self) -> None: + events: list[str] = [] + batch = make_batch() + + async def lock(*_: object) -> SimpleNamespace: + events.append('lock') + return batch + + async def validate(*_: object) -> list[int]: + events.append('validate') + return [2, 3] + + async def replace(*_: object) -> None: + events.append('replace') + + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(side_effect=lock)) as lock_batch, + patch.object(TestBatchDao, 'list_active_user_ids', AsyncMock(side_effect=validate)) as active_users, + patch.object(TestBatchDao, 'replace_batch_users', AsyncMock(side_effect=replace)) as replace_users, + ): + result = await TestBatchService.replace_users( + self.db, + 3, + TestBatchUserAssignModel(userIds=[3, 2, 3]), + self.admin, + ) + + self.assertEqual(['lock', 'validate', 'replace'], events) + lock_batch.assert_awaited_once_with(self.db, 3) + active_users.assert_awaited_once_with(self.db, [3, 2]) + replace_users.assert_awaited_once() + self.assertEqual((self.db, 3, [3, 2], 'admin'), replace_users.await_args.args[:4]) + self.assertEqual('测试批次用户授权成功', result.message) + self.db.commit.assert_awaited_once_with() + + async def test_replace_users_accepts_empty_assignment_without_user_query(self) -> None: + active_users = AsyncMock() + replace_users = AsyncMock() + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=make_batch())), + patch.object(TestBatchDao, 'list_active_user_ids', active_users), + patch.object(TestBatchDao, 'replace_batch_users', replace_users), + ): + await TestBatchService.replace_users( + self.db, + 3, + TestBatchUserAssignModel(userIds=[]), + self.admin, + ) + + active_users.assert_not_awaited() + self.assertEqual((self.db, 3, [], 'admin'), replace_users.await_args.args[:4]) + self.db.commit.assert_awaited_once_with() + + async def test_replace_users_rejects_missing_disabled_or_deleted_user(self) -> None: + replace_users = AsyncMock() + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=make_batch())), + patch.object(TestBatchDao, 'list_active_user_ids', AsyncMock(return_value=[2])), + patch.object(TestBatchDao, 'replace_batch_users', replace_users), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.replace_users( + self.db, + 3, + TestBatchUserAssignModel(userIds=[2, 9]), + self.admin, + ) + + self.assertIn('不存在、停用或已删除', raised.exception.message) + replace_users.assert_not_awaited() + self.db.rollback.assert_awaited_once_with() + + async def test_replace_users_rejects_non_active_batch_after_lock(self) -> None: + for status, message in (('CLOSED', '已关闭'), ('PURGED', '已清理'), ('UNKNOWN', '状态异常')): + with self.subTest(status=status): + self.db.reset_mock() + active_users = AsyncMock() + replace_users = AsyncMock() + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=make_batch(status=status))), + patch.object(TestBatchDao, 'list_active_user_ids', active_users), + patch.object(TestBatchDao, 'replace_batch_users', replace_users), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.replace_users( + self.db, + 3, + TestBatchUserAssignModel(userIds=[2]), + self.admin, + ) + + self.assertIn(message, raised.exception.message) + active_users.assert_not_awaited() + replace_users.assert_not_awaited() + self.db.rollback.assert_awaited_once_with() + + async def test_replace_users_integrity_failure_rolls_back(self) -> None: + error = IntegrityError('insert', {}, RuntimeError('foreign key')) + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=make_batch())), + patch.object(TestBatchDao, 'list_active_user_ids', AsyncMock(return_value=[2])), + patch.object(TestBatchDao, 'replace_batch_users', AsyncMock(side_effect=error)), + self.assertRaises(ServiceException) as raised, + ): + await TestBatchService.replace_users( + self.db, + 3, + TestBatchUserAssignModel(userIds=[2]), + self.admin, + ) + + self.assertIn('授权失败', raised.exception.message) + self.db.rollback.assert_awaited_once_with() + self.db.commit.assert_not_awaited() + + async def test_only_active_batch_can_close_and_preview_is_invalidated(self) -> None: + batch = make_batch() + with patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)) as lock_batch: + result = await TestBatchService.close_batch(self.db, 3, self.admin) + + lock_batch.assert_awaited_once_with(self.db, 3) + self.assertEqual('CLOSED', batch.status) + self.assertEqual(1, batch.closed_by) + self.assertIsInstance(batch.closed_at, datetime) + self.assertIsNone(batch.preview_action) + self.assertIsNone(batch.preview_token_hash) + self.assertIsNone(batch.preview_scope_hash) + self.assertIsNone(batch.preview_expires_at) + self.assertEqual('测试批次关闭成功', result.message) + self.db.commit.assert_awaited_once_with() + self.db.rollback.assert_not_awaited() + + async def test_close_missing_or_illegal_state_rolls_back(self) -> None: + cases = ( + (None, '不存在'), + (make_batch(status='CLOSED'), '已关闭'), + (make_batch(status='PURGED'), '已清理'), + (make_batch(status='UNKNOWN'), '状态异常'), + ) + for batch, message in cases: + with self.subTest(message=message): + self.db.reset_mock() + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.close_batch(self.db, 3, self.admin) + + self.assertIn(message, raised.exception.message) + self.db.rollback.assert_awaited_once_with() + self.db.commit.assert_not_awaited() + + async def test_closed_batch_can_reopen_and_close_metadata_is_cleared(self) -> None: + batch = make_batch(status='CLOSED') + batch.closed_by = 1 + batch.closed_at = datetime(2026, 9, 22, 11, 0, 0) + with patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)) as lock_batch: + result = await TestBatchService.reopen_batch(self.db, 3, self.admin) + + lock_batch.assert_awaited_once_with(self.db, 3) + self.assertEqual('ACTIVE', batch.status) + self.assertIsNone(batch.closed_by) + self.assertIsNone(batch.closed_at) + self.assertIsNone(batch.preview_action) + self.assertIsNone(batch.preview_token_hash) + self.assertIsNone(batch.preview_scope_hash) + self.assertIsNone(batch.preview_expires_at) + self.assertEqual('测试批次重新开启成功', result.message) + self.db.commit.assert_awaited_once_with() + + async def test_reopen_missing_or_illegal_state_rolls_back(self) -> None: + cases = ( + (None, '不存在'), + (make_batch(status='ACTIVE'), '已处于启用状态'), + (make_batch(status='PURGED'), '已清理'), + (make_batch(status='UNKNOWN'), '状态异常'), + ) + for batch, message in cases: + with self.subTest(message=message): + self.db.reset_mock() + with ( + patch.object(TestBatchDao, 'get_batch_for_update', AsyncMock(return_value=batch)), + self.assertRaises(ServiceWarning) as raised, + ): + await TestBatchService.reopen_batch(self.db, 3, self.admin) + + self.assertIn(message, raised.exception.message) + self.db.rollback.assert_awaited_once_with() + self.db.commit.assert_not_awaited() + + +class TestBatchDaoLifecycleTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.db = AsyncMock(spec=AsyncSession) + + async def test_add_batch_only_adds_without_committing(self) -> None: + batch = BizTestBatch(batch_code='TB-1', batch_name='测试', status='ACTIVE') + + await TestBatchDao.add_batch(self.db, batch) + + self.db.add.assert_called_once_with(batch) + self.db.commit.assert_not_awaited() + + async def test_list_batches_orders_newest_first_without_status_filter(self) -> None: + expected = [make_batch(batch_id=4), make_batch(batch_id=3, status='CLOSED')] + result_proxy = MagicMock() + result_proxy.scalars.return_value.all.return_value = expected + self.db.execute.return_value = result_proxy + + result = await TestBatchDao.list_batches(self.db) + + self.assertEqual(expected, result) + sql = compile_sql(self.db.execute.await_args.args[0]) + self.assertIn('ORDER BY biz_test_batch.batch_id DESC', sql) + self.assertNotIn('WHERE', sql) + self.db.commit.assert_not_awaited() + + async def test_active_user_lookup_filters_disabled_and_deleted_users(self) -> None: + result_proxy = MagicMock() + result_proxy.scalars.return_value.all.return_value = [2, 3] + self.db.execute.return_value = result_proxy + + result = await TestBatchDao.list_active_user_ids(self.db, [2, 3, 9]) + + self.assertEqual([2, 3], result) + sql = compile_sql(self.db.execute.await_args.args[0]) + self.assertIn('sys_user.user_id IN (2, 3, 9)', sql) + self.assertIn("sys_user.status = '0'", sql) + self.assertIn("sys_user.del_flag = '0'", sql) + + async def test_empty_active_user_lookup_skips_database(self) -> None: + self.assertEqual([], await TestBatchDao.list_active_user_ids(self.db, [])) + self.db.execute.assert_not_awaited() + + async def test_replace_batch_users_deletes_then_adds_without_commit(self) -> None: + created_at = datetime(2026, 9, 22, 12, 0, 0) + + await TestBatchDao.replace_batch_users(self.db, 3, [8, 9], 'admin', created_at) + + delete_sql = compile_sql(self.db.execute.await_args.args[0]) + self.assertIn('DELETE FROM biz_test_batch_user', delete_sql) + self.assertIn('biz_test_batch_user.batch_id = 3', delete_sql) + added = self.db.add_all.call_args.args[0] + self.assertEqual(2, len(added)) + self.assertTrue(all(isinstance(item, BizTestBatchUser) for item in added)) + self.assertEqual([(3, 8, 'admin'), (3, 9, 'admin')], [(item.batch_id, item.user_id, item.create_by) for item in added]) + self.assertTrue(all(item.create_time == created_at for item in added)) + self.db.commit.assert_not_awaited() + + async def test_count_owned_rows_queries_exactly_sixteen_scoped_tables(self) -> None: + results = [] + for index in range(len(OWNED_TABLES)): + result_proxy = MagicMock() + result_proxy.scalar.return_value = index + results.append(result_proxy) + self.db.execute.side_effect = results + + counts = await TestBatchDao.count_owned_rows(self.db, 7) + + self.assertEqual({name: index for index, (name, _) in enumerate(OWNED_TABLES)}, counts) + self.assertEqual(16, self.db.execute.await_count) + statements = [compile_sql(call.args[0]) for call in self.db.execute.await_args_list] + for statement, (table_name, _) in zip(statements, OWNED_TABLES, strict=True): + self.assertIn(f'FROM {table_name}', statement) + self.assertIn(f'{table_name}.test_batch_id = 7', statement) + self.db.commit.assert_not_awaited() + + +if __name__ == '__main__': + unittest.main()