import inspect import unittest from contextlib import ExitStack from datetime import datetime from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock, patch from fastapi import FastAPI from fastapi.routing import APIRoute from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from common.aspect.interface_auth import CheckUserInterfaceAuth from common.aspect.pre_auth import PreAuth from common.vo import CrudResponseModel, DataResponseModel, PageModel, PageResponseModel from exceptions.exception import PermissionException, ServiceException, ServiceWarning from module_admin.service.login_service import LoginService from module_business.controller import ( customer_controller as customer_controller_module, ) from module_business.controller import line_controller as line_controller_module from module_business.controller import site_controller as site_controller_module from module_business.controller import strain_controller as strain_controller_module from module_business.controller import warehouse_controller as warehouse_controller_module from module_business.dao.master_dao import MasterDao from module_business.dependency.test_batch_dependency import TestBatchScopeResolver from module_business.domain.test_batch_scope import TestBatchScope from module_business.entity.do.master_do import ( BizCustomer, BizCustomerSite, BizProductionLine, BizStrain, BizWarehouse, ) from module_business.entity.vo.master_vo import ( CustomerModel, CustomerPageQueryModel, DeleteCustomerModel, DeleteLineModel, DeleteSiteModel, DeleteStrainModel, DeleteWarehouseModel, LineModel, LinePageQueryModel, SiteModel, SitePageQueryModel, StrainModel, StrainPageQueryModel, WarehouseModel, WarehousePageQueryModel, ) from module_business.service.master_service import MasterService from module_business.service.test_batch_access_service import TestBatchAccessService from tests.module_business.test_batch_fixtures import make_admin, make_current_user, make_test_scope from utils.page_util import PageUtil RESOURCE_IDS = { 'strain': 'strain_id', 'customer': 'customer_id', 'site': 'site_id', 'line': 'line_id', 'warehouse': 'warehouse_id', } def make_models(resource_id: int | None = None) -> dict[str, object]: return { 'strain': StrainModel(strainId=resource_id, strainCode='S001', strainName='测试菌种'), 'customer': CustomerModel(customerId=resource_id, customerCode='C001', customerName='测试客户'), 'site': SiteModel( siteId=resource_id, customerId=11, siteCode='SITE001', siteName='测试现场', ), 'line': LineModel( lineId=resource_id, siteId=22, lineCode='L001', lineName='测试产线', ), 'warehouse': WarehouseModel( warehouseId=resource_id, warehouseCode='W001', warehouseName='测试仓库', ), } def compile_sql(statement: Any) -> str: return ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split()) def integrity_error() -> IntegrityError: return IntegrityError('statement', {}, RuntimeError('duplicate')) class TestMasterService(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) self.admin = make_admin() self.scope = TestBatchScope.formal() self.strain = StrainModel(strainCode='S001', strainName='测试菌种') self.line = LineModel(siteId=1, lineCode='L001', lineName='测试产线') def _patch_enabled_parent(self, stack: ExitStack, resource: str) -> None: if resource == 'site': stack.enter_context( patch.object( MasterDao, 'get_customer', new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=None)), ) ) elif resource == 'line': stack.enter_context( patch.object( MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=None)), ) ) async def test_formal_lists_return_models_and_forward_scope_for_all_resources(self) -> None: for resource in RESOURCE_IDS: rows = [make_models(1)[resource], make_models(2)[resource]] page = PageModel(rows=rows, pageNum=1, pageSize=10, total=2, hasNext=False) list_mock = AsyncMock(return_value=page) query = MagicMock() with ( self.subTest(resource=resource), patch.object(MasterDao, f'list_{resource}', new=list_mock), ): result = await getattr(MasterService, f'list_{resource}')( self.db, query, self.scope, ) self.assertTrue(all(isinstance(row, type(rows[0])) for row in result.rows)) list_mock.assert_awaited_once_with(self.db, query, self.scope, True) async def test_duplicate_strain_code_is_rejected(self) -> None: with ( patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=object())), self.assertRaises(ServiceException), ): await MasterService.add_strain(self.db, self.strain, self.admin, self.scope) async def test_line_must_belong_to_enabled_site(self) -> None: with ( patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=None)), self.assertRaises(ServiceException), ): await MasterService.add_line(self.db, self.line, self.admin, self.scope) async def test_duplicate_code_is_rejected_for_all_five_resources(self) -> None: for resource, model in make_models().items(): with self.subTest(resource=resource), ExitStack() as stack: self._patch_enabled_parent(stack, resource) stack.enter_context( patch.object( MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=SimpleNamespace()), ) ) with self.assertRaises(ServiceException): await getattr(MasterService, f'add_{resource}')( self.db, model, self.admin, self.scope, ) async def test_edit_unique_check_excludes_current_record_for_all_resources(self) -> None: for resource, model in make_models(7).items(): id_field = RESOURCE_IDS[resource] current = SimpleNamespace(**{id_field: 7}, test_batch_id=None) with self.subTest(resource=resource), ExitStack() as stack: self.db.reset_mock() self._patch_enabled_parent(stack, resource) stack.enter_context(patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=current))) stack.enter_context( patch.object(MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=current)) ) update_mock = stack.enter_context(patch.object(MasterDao, f'update_{resource}', new=AsyncMock())) result = await getattr(MasterService, f'edit_{resource}')( self.db, model, self.admin, self.scope, ) self.assertIsInstance(result, CrudResponseModel) self.assertTrue(result.is_success) update_mock.assert_awaited_once() self.db.commit.assert_awaited_once() async def test_edit_duplicate_code_from_another_record_is_rejected_for_all_resources(self) -> None: for resource, model in make_models(7).items(): id_field = RESOURCE_IDS[resource] with self.subTest(resource=resource), ExitStack() as stack: self._patch_enabled_parent(stack, resource) stack.enter_context( patch.object( MasterDao, f'get_{resource}', new=AsyncMock(return_value=SimpleNamespace(**{id_field: 7}, test_batch_id=None)), ) ) stack.enter_context( patch.object( MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=SimpleNamespace(**{id_field: 8})), ) ) with self.assertRaises(ServiceException): await getattr(MasterService, f'edit_{resource}')( self.db, model, self.admin, self.scope, ) async def test_site_add_requires_existing_enabled_customer(self) -> None: site = make_models()['site'] for parent in (None, SimpleNamespace(status='1', test_batch_id=None)): with ( self.subTest(parent=parent), patch.object(MasterDao, 'get_customer', new=AsyncMock(return_value=parent)), self.assertRaises(ServiceException), ): await MasterService.add_site(self.db, site, self.admin, self.scope) with ( patch.object( MasterDao, 'get_customer', new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=None)), ), patch.object(MasterDao, 'find_site_by_code', new=AsyncMock(return_value=None)), patch.object(MasterDao, 'add_site', new=AsyncMock()), ): result = await MasterService.add_site(self.db, site, self.admin, self.scope) self.assertTrue(result.is_success) async def test_line_add_requires_existing_enabled_site(self) -> None: line = make_models()['line'] for parent in (None, SimpleNamespace(status='1', test_batch_id=None)): with ( self.subTest(parent=parent), patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=parent)), self.assertRaises(ServiceException), ): await MasterService.add_line(self.db, line, self.admin, self.scope) with ( patch.object( MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=None)), ), patch.object(MasterDao, 'find_line_by_code', new=AsyncMock(return_value=None)), patch.object(MasterDao, 'add_line', new=AsyncMock()), ): result = await MasterService.add_line(self.db, line, self.admin, self.scope) self.assertTrue(result.is_success) async def test_site_and_line_edit_require_enabled_target_parent(self) -> None: cases = ( ('site', make_models(7)['site'], 'get_customer'), ('line', make_models(7)['line'], 'get_site'), ) for resource, model, parent_method in cases: for parent in (None, SimpleNamespace(status='1', test_batch_id=None)): with self.subTest(resource=resource, parent=parent), ExitStack() as stack: stack.enter_context( patch.object( MasterDao, f'get_{resource}', new=AsyncMock( return_value=SimpleNamespace( **{RESOURCE_IDS[resource]: 7}, test_batch_id=None, ) ), ) ) stack.enter_context(patch.object(MasterDao, parent_method, new=AsyncMock(return_value=parent))) with self.assertRaises(ServiceException): await getattr(MasterService, f'edit_{resource}')( self.db, model, self.admin, self.scope, ) with self.subTest(resource=resource, parent='enabled'), ExitStack() as stack: stack.enter_context( patch.object( MasterDao, f'get_{resource}', new=AsyncMock( return_value=SimpleNamespace( **{RESOURCE_IDS[resource]: 7}, test_batch_id=None, ) ), ) ) stack.enter_context( patch.object( MasterDao, parent_method, new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=None)), ) ) stack.enter_context( patch.object(MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=None)) ) stack.enter_context(patch.object(MasterDao, f'update_{resource}', new=AsyncMock())) result = await getattr(MasterService, f'edit_{resource}')( self.db, model, self.admin, self.scope, ) self.assertTrue(result.is_success) async def test_detail_missing_is_explicit_for_all_resources(self) -> None: for resource in RESOURCE_IDS: with ( self.subTest(resource=resource), patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=None)), self.assertRaises(ServiceException), ): await getattr(MasterService, f'detail_{resource}')(self.db, 99, self.scope) class TestMasterBatchScopeService(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) self.admin = make_admin() self.scope = make_test_scope(9) self.formal_scope = TestBatchScope.formal() async def test_lists_and_details_forward_trusted_scope_to_dao(self) -> None: for resource in RESOURCE_IDS: query = MagicMock() page = PageModel(rows=[], pageNum=1, pageSize=10, total=0, hasNext=False) list_mock = AsyncMock(return_value=page) detail_model = make_models(7)[resource] detail_record = SimpleNamespace(**detail_model.model_dump(), test_batch_id=9) detail_mock = AsyncMock(return_value=detail_record) with ( self.subTest(resource=resource), patch.object(MasterDao, f'list_{resource}', new=list_mock), patch.object(MasterDao, f'get_{resource}', new=detail_mock), ): listed = await getattr(MasterService, f'list_{resource}')( self.db, query, self.scope, is_page=True, ) detailed = await getattr(MasterService, f'detail_{resource}')(self.db, 7, self.scope) self.assertIsInstance(listed, PageModel) self.assertEqual(detail_model, detailed) list_mock.assert_awaited_once_with(self.db, query, self.scope, True) detail_mock.assert_awaited_once_with(self.db, 7, self.scope) async def test_test_scope_add_locks_batch_before_any_master_query_and_inherits_batch(self) -> None: for resource, model in make_models().items(): events: list[str] = [] lock_batch = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('lock-batch')) find_duplicate = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('find-duplicate')) add_record = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('add-record')) parent_method = None parent_lookup = None if resource == 'site': parent_method = 'get_customer' elif resource == 'line': parent_method = 'get_site' with self.subTest(resource=resource), ExitStack() as stack: self.db.reset_mock() stack.enter_context(patch.object(TestBatchAccessService, 'lock_active_batch', new=lock_batch)) stack.enter_context(patch.object(MasterDao, f'find_{resource}_by_code', new=find_duplicate)) stack.enter_context(patch.object(MasterDao, f'add_{resource}', new=add_record)) if parent_method is not None: parent_lookup = AsyncMock( side_effect=lambda *_args, _events=events: ( _events.append('get-parent') or SimpleNamespace(status='0', test_batch_id=9) ) ) stack.enter_context(patch.object(MasterDao, parent_method, new=parent_lookup)) result = await getattr(MasterService, f'add_{resource}')( self.db, model, self.admin, self.scope, ) self.assertTrue(result.is_success) self.assertEqual('lock-batch', events[0]) lock_batch.assert_awaited_once_with(self.db, self.scope) self.assertEqual(9, add_record.await_args.args[2]) if parent_lookup is not None: parent_lookup.assert_awaited_once() async def test_test_scope_edit_locks_batch_before_business_rows_and_preserves_scope(self) -> None: for resource, model in make_models(7).items(): events: list[str] = [] id_field = RESOURCE_IDS[resource] current = SimpleNamespace(**{id_field: 7}, test_batch_id=9) lock_batch = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('lock-batch')) get_current = AsyncMock( side_effect=lambda *_args, _events=events, _current=current, **_kwargs: ( _events.append('get-current') or _current ) ) update_record = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('update-record')) with self.subTest(resource=resource), ExitStack() as stack: self.db.reset_mock() stack.enter_context(patch.object(TestBatchAccessService, 'lock_active_batch', new=lock_batch)) stack.enter_context(patch.object(MasterDao, f'get_{resource}', new=get_current)) stack.enter_context( patch.object(MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=current)) ) stack.enter_context(patch.object(MasterDao, f'update_{resource}', new=update_record)) if resource == 'site': stack.enter_context( patch.object( MasterDao, 'get_customer', new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=9)), ) ) elif resource == 'line': stack.enter_context( patch.object( MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=9)), ) ) result = await getattr(MasterService, f'edit_{resource}')( self.db, model, self.admin, self.scope, ) self.assertTrue(result.is_success) self.assertEqual('lock-batch', events[0]) get_current.assert_awaited_once_with(self.db, 7, self.scope, for_update=True) update_record.assert_awaited_once_with(self.db, model, self.scope) async def test_regular_user_cannot_add_or_edit_test_master_before_database_access(self) -> None: current_user = make_current_user(8) lock_batch = AsyncMock() dao_call = AsyncMock() with ( patch.object(TestBatchAccessService, 'lock_active_batch', new=lock_batch), patch.object(MasterDao, 'find_strain_by_code', new=dao_call), patch.object(MasterDao, 'get_strain', new=dao_call), ): with self.assertRaisesRegex(PermissionException, '仅超级管理员'): await MasterService.add_strain( self.db, make_models()['strain'], current_user, self.scope, ) with self.assertRaisesRegex(PermissionException, '仅超级管理员'): await MasterService.edit_strain( self.db, make_models(7)['strain'], current_user, self.scope, ) lock_batch.assert_not_awaited() dao_call.assert_not_awaited() async def test_closed_batch_rejection_happens_before_unique_parent_or_business_queries(self) -> None: closed = PermissionException('', '当前测试批次已关闭') dao_call = AsyncMock() with ( patch.object(TestBatchAccessService, 'lock_active_batch', new=AsyncMock(side_effect=closed)), patch.object(MasterDao, 'get_customer', new=dao_call), patch.object(MasterDao, 'find_site_by_code', new=dao_call), patch.object(MasterDao, 'add_site', new=dao_call), self.assertRaisesRegex(PermissionException, '已关闭'), ): await MasterService.add_site( self.db, make_models()['site'], self.admin, self.scope, ) dao_call.assert_not_awaited() async def test_parent_relations_reject_formal_test_and_cross_batch_mixing(self) -> None: cases = ( ('site', 'get_customer', TestBatchScope.formal(), 9), ('site', 'get_customer', self.scope, None), ('site', 'get_customer', self.scope, 10), ('line', 'get_site', TestBatchScope.formal(), 9), ('line', 'get_site', self.scope, None), ('line', 'get_site', self.scope, 10), ) for resource, parent_method, scope, parent_batch_id in cases: with self.subTest(resource=resource, scope=scope, parent_batch_id=parent_batch_id), ExitStack() as stack: stack.enter_context(patch.object(TestBatchAccessService, 'lock_active_batch', new=AsyncMock())) stack.enter_context( patch.object( MasterDao, parent_method, new=AsyncMock(return_value=SimpleNamespace(status='0', test_batch_id=parent_batch_id)), ) ) stack.enter_context( patch.object(MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=None)) ) stack.enter_context(patch.object(MasterDao, f'add_{resource}', new=AsyncMock())) with self.assertRaisesRegex(ServiceException, '不属于当前数据模式或测试批次'): await getattr(MasterService, f'add_{resource}')( self.db, make_models()[resource], self.admin, scope, ) async def test_all_legacy_master_deletes_are_rejected_before_delete_dao(self) -> None: query_result = MagicMock() query_result.scalar.return_value = 0 query_result.scalar_one_or_none.return_value = SimpleNamespace() self.db.execute = AsyncMock(return_value=query_result) for resource in RESOURCE_IDS: delete_dao = AsyncMock() with self.subTest(resource=resource): with ( patch.object(MasterDao, f'delete_{resource}', new=delete_dao, create=True), self.assertRaisesRegex(ServiceWarning, '不支持单条物理删除，请通过测试数据批次清理'), ): await getattr(MasterService, f'remove_{resource}')(self.db, 1) delete_dao.assert_not_awaited() self.db.commit.assert_not_awaited() async def test_add_ignores_client_primary_key_before_calling_dao(self) -> None: model = make_models(77)['strain'] with ( patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(MasterDao, 'add_strain', new=AsyncMock()) as add_mock, ): await MasterService.add_strain(self.db, model, self.admin, self.formal_scope) persisted_model = add_mock.await_args.args[1] self.assertIsNone(persisted_model.strain_id) self.assertIsNone(add_mock.await_args.args[2]) async def test_integrity_errors_from_add_update_and_commit_are_stable(self) -> None: for phase in ('add', 'update', 'commit'): db = MagicMock(spec=AsyncSession) with self.subTest(phase=phase), ExitStack() as stack: if phase in {'add', 'commit'}: stack.enter_context( patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)) ) stack.enter_context( patch.object( MasterDao, 'add_strain', new=AsyncMock(side_effect=integrity_error() if phase == 'add' else None), ) ) if phase == 'commit': db.commit.side_effect = integrity_error() call = MasterService.add_strain( db, make_models()['strain'], self.admin, self.formal_scope, ) else: stack.enter_context( patch.object( MasterDao, 'get_strain', new=AsyncMock(return_value=SimpleNamespace(strain_id=7, test_batch_id=None)), ) ) stack.enter_context( patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)) ) stack.enter_context( patch.object(MasterDao, 'update_strain', new=AsyncMock(side_effect=integrity_error())) ) call = MasterService.edit_strain( db, make_models(7)['strain'], self.admin, self.formal_scope, ) with self.assertRaises(ServiceException) as raised: await call self.assertIsNotNone(raised.exception.message) self.assertNotIn('duplicate', raised.exception.message) db.rollback.assert_awaited_once() if phase != 'commit': db.commit.assert_not_awaited() async def test_non_integrity_exception_rolls_back_and_is_reraised(self) -> None: error = RuntimeError('unexpected') with ( patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(MasterDao, 'add_strain', new=AsyncMock(side_effect=error)), self.assertRaises(RuntimeError) as raised, ): await MasterService.add_strain( self.db, make_models()['strain'], self.admin, self.formal_scope, ) self.assertIs(error, raised.exception) self.db.rollback.assert_awaited_once() self.db.commit.assert_not_awaited() class TestMasterDao(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) async def test_list_queries_cover_vo_filters_and_stable_primary_key_order(self) -> None: cases = ( ( 'strain', StrainPageQueryModel(strainCode='S001', strainName='菌种', status='0', pageNum=2, pageSize=5), ('biz_strain.strain_code LIKE', 'biz_strain.strain_name LIKE', 'biz_strain.status =', 'strain_id'), ), ( 'customer', CustomerPageQueryModel(customerCode='C001', customerName='客户', status='0', pageNum=2, pageSize=5), ( 'biz_customer.customer_code LIKE', 'biz_customer.customer_name LIKE', 'biz_customer.status =', 'customer_id', ), ), ( 'site', SitePageQueryModel( customerId=11, siteCode='SITE001', siteName='现场', status='0', pageNum=2, pageSize=5, ), ( 'biz_customer_site.customer_id = 11', 'biz_customer_site.site_code LIKE', 'biz_customer_site.site_name LIKE', 'biz_customer_site.status =', 'site_id', ), ), ( 'line', LinePageQueryModel(siteId=22, lineCode='L001', lineName='产线', status='0', pageNum=2, pageSize=5), ( 'biz_production_line.site_id = 22', 'biz_production_line.line_code LIKE', 'biz_production_line.line_name LIKE', 'biz_production_line.status =', 'line_id', ), ), ( 'warehouse', WarehousePageQueryModel(warehouseCode='W001', warehouseName='仓库', status='0', pageNum=2, pageSize=5), ( 'biz_warehouse.warehouse_code LIKE', 'biz_warehouse.warehouse_name LIKE', 'biz_warehouse.status =', 'warehouse_id', ), ), ) page = PageModel(rows=[], pageNum=2, pageSize=5, total=0, hasNext=False) for resource, query_model, expected_tokens in cases: with ( self.subTest(resource=resource), patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=page)) as paginate_mock, ): result = await getattr(MasterDao, f'list_{resource}')( self.db, query_model, TestBatchScope.formal(), is_page=True, ) statement = paginate_mock.await_args.args[1] sql = compile_sql(statement) self.assertIs(page, result) for token in expected_tokens[:-1]: self.assertIn(token, sql) self.assertIn(f'ORDER BY {statement.get_final_froms()[0].name}.{expected_tokens[-1]}', sql) self.assertEqual((2, 5, True), paginate_mock.await_args.args[2:]) async def test_strain_code_search_keeps_partial_matching_for_leading_zero_codes(self) -> None: page = PageModel(rows=[], pageNum=1, pageSize=10, total=0, hasNext=False) for strain_code in ('01', '1'): with ( self.subTest(strain_code=strain_code), patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=page)) as paginate_mock, ): result = await MasterDao.list_strain( self.db, StrainPageQueryModel(strainCode=strain_code, pageNum=1, pageSize=10), TestBatchScope.formal(), is_page=True, ) statement = paginate_mock.await_args.args[1] self.assertIs(page, result) self.assertIn(f"biz_strain.strain_code LIKE '%{strain_code}%'", compile_sql(statement)) async def test_get_queries_use_expected_table_and_primary_key(self) -> None: cases = ( ('strain', 'biz_strain', 'strain_id'), ('customer', 'biz_customer', 'customer_id'), ('site', 'biz_customer_site', 'site_id'), ('line', 'biz_production_line', 'line_id'), ('warehouse', 'biz_warehouse', 'warehouse_id'), ) for resource, table, id_field in cases: result = MagicMock() result.scalar_one_or_none.return_value = None self.db.execute = AsyncMock(return_value=result) await getattr(MasterDao, f'get_{resource}')( self.db, 9, TestBatchScope.formal(), ) sql = compile_sql(self.db.execute.await_args.args[0]) self.assertIn(f'FROM {table}', sql) self.assertIn(f'{table}.{id_field} = 9', sql) async def test_locking_master_gets_use_current_reads_and_refresh_identity_map(self) -> None: for resource in ('warehouse', 'strain', 'line'): with self.subTest(resource=resource): lock_parameter = inspect.signature(getattr(MasterDao, f'get_{resource}')).parameters.get('for_update') self.assertIsNotNone(lock_parameter) self.assertIs(inspect.Parameter.KEYWORD_ONLY, lock_parameter.kind) self.assertFalse(lock_parameter.default) result = MagicMock() result.scalar_one_or_none.return_value = None self.db.execute = AsyncMock(return_value=result) await getattr(MasterDao, f'get_{resource}')( self.db, 9, TestBatchScope.formal(), for_update=True, ) statement = self.db.execute.await_args.args[0] self.assertIn('FOR UPDATE', compile_sql(statement)) self.assertIs(True, statement.get_execution_options().get('populate_existing')) async def test_unique_queries_use_global_or_parent_scoped_predicates(self) -> None: cases = ( ('strain', ('S001',), 'biz_strain', ('strain_code',)), ('customer', ('C001',), 'biz_customer', ('customer_code',)), ('warehouse', ('W001',), 'biz_warehouse', ('warehouse_code',)), ('site', ('SITE001', 11), 'biz_customer_site', ('site_code', 'customer_id')), ('line', ('L001', 22), 'biz_production_line', ('line_code', 'site_id')), ) for resource, args, table, predicates in cases: result = MagicMock() result.scalar_one_or_none.return_value = None self.db.execute = AsyncMock(return_value=result) await getattr(MasterDao, f'find_{resource}_by_code')(self.db, *args) statement = self.db.execute.await_args.args[0] sql = compile_sql(statement) self.assertIn(f'FROM {table}', sql) self.assertNotIn('test_batch_id', compile_sql(statement.whereclause)) for predicate in predicates: self.assertIn(f'{table}.{predicate} =', sql) async def test_add_ignores_client_primary_keys_flushes_and_never_commits(self) -> None: expected_classes = { 'strain': BizStrain, 'customer': BizCustomer, 'site': BizCustomerSite, 'line': BizProductionLine, 'warehouse': BizWarehouse, } for resource, model in make_models(77).items(): with self.subTest(resource=resource): self.db.reset_mock() record = await getattr(MasterDao, f'add_{resource}')(self.db, model, None) self.assertIsInstance(record, expected_classes[resource]) self.assertIsNone(getattr(record, RESOURCE_IDS[resource])) self.assertIsNone(record.test_batch_id) self.db.add.assert_called_once_with(record) self.db.flush.assert_awaited_once() self.db.commit.assert_not_awaited() self.db.rollback.assert_not_awaited() async def test_update_statements_use_primary_key_scope_and_preserve_creation_audit(self) -> None: table_names = { 'strain': 'biz_strain', 'customer': 'biz_customer', 'site': 'biz_customer_site', 'line': 'biz_production_line', 'warehouse': 'biz_warehouse', } created_at = datetime(2026, 1, 2, 3, 4, 5) for resource, model in make_models(7).items(): model.create_by = '客户端伪造创建人' model.create_time = created_at model.update_by = 'server-user' with self.subTest(resource=resource): self.db.reset_mock() await getattr(MasterDao, f'update_{resource}')( self.db, model, TestBatchScope.formal(), ) update_sql = compile_sql(self.db.execute.await_args.args[0]) table = table_names[resource] id_field = RESOURCE_IDS[resource] self.assertIn(f'UPDATE {table} SET', update_sql) self.assertIn(f'WHERE {table}.{id_field} = 7', update_sql) self.assertIn(f'{table}.test_batch_id IS NULL', update_sql) self.assertNotIn('create_by', update_sql) self.assertNotIn('create_time', update_sql) self.db.commit.assert_not_awaited() class TestMasterDaoBatchScope(unittest.IsolatedAsyncioTestCase): TABLES = { 'strain': ('biz_strain', 'strain_id'), 'customer': ('biz_customer', 'customer_id'), 'site': ('biz_customer_site', 'site_id'), 'line': ('biz_production_line', 'line_id'), 'warehouse': ('biz_warehouse', 'warehouse_id'), } def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) async def test_all_master_lists_apply_formal_and_test_scope_in_sql(self) -> None: queries = { 'strain': StrainPageQueryModel(), 'customer': CustomerPageQueryModel(), 'site': SitePageQueryModel(), 'line': LinePageQueryModel(), 'warehouse': WarehousePageQueryModel(), } for resource, (table, _) in self.TABLES.items(): for scope, expected in ( (TestBatchScope.formal(), f'{table}.test_batch_id IS NULL'), (make_test_scope(9), f'{table}.test_batch_id = 9'), ): with ( self.subTest(resource=resource, scope=scope.mode), patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=[])) as paginate, ): await getattr(MasterDao, f'list_{resource}')( self.db, queries[resource], scope, is_page=False, ) self.assertIn(expected, compile_sql(paginate.await_args.args[1])) async def test_all_master_details_apply_scope_and_keep_locking(self) -> None: query_result = MagicMock() query_result.scalar_one_or_none.return_value = None self.db.execute = AsyncMock(return_value=query_result) for resource, (table, id_field) in self.TABLES.items(): for scope, expected in ( (TestBatchScope.formal(), f'{table}.test_batch_id IS NULL'), (make_test_scope(9), f'{table}.test_batch_id = 9'), ): with self.subTest(resource=resource, scope=scope.mode): self.db.execute.reset_mock() await getattr(MasterDao, f'get_{resource}')( self.db, 7, scope, for_update=True, ) sql = compile_sql(self.db.execute.await_args.args[0]) self.assertIn(f'{table}.{id_field} = 7', sql) self.assertIn(expected, sql) self.assertIn('FOR UPDATE', sql) async def test_add_receives_batch_id_separately_and_update_cannot_change_it(self) -> None: table_by_resource = {resource: table for resource, (table, _) in self.TABLES.items()} for resource, model in make_models(7).items(): with self.subTest(resource=resource): self.db.reset_mock() record = await getattr(MasterDao, f'add_{resource}')(self.db, model, 9) self.assertEqual(9, record.test_batch_id) self.db.reset_mock() await getattr(MasterDao, f'update_{resource}')(self.db, model, make_test_scope(9)) sql = compile_sql(self.db.execute.await_args.args[0]) self.assertIn(f'{table_by_resource[resource]}.test_batch_id = 9', sql) self.assertNotIn('SET test_batch_id=', sql) class TestMasterControllers(unittest.TestCase): ROUTERS = { 'strain': strain_controller_module.strain_controller, 'customer': customer_controller_module.customer_controller, 'site': site_controller_module.site_controller, 'line': line_controller_module.line_controller, 'warehouse': warehouse_controller_module.warehouse_controller, } MODULES = { 'strain': strain_controller_module, 'customer': customer_controller_module, 'site': site_controller_module, 'line': line_controller_module, 'warehouse': warehouse_controller_module, } def test_openapi_contains_all_resource_paths_methods_and_response_models(self) -> None: app = FastAPI() for router in self.ROUTERS.values(): app.include_router(router) schema = app.openapi() for resource in self.ROUTERS: prefix = f'/master/{resource}' self.assertIn(f'{prefix}/list', schema['paths']) self.assertIn('get', schema['paths'][f'{prefix}/list']) self.assertIn(prefix, schema['paths']) self.assertIn('post', schema['paths'][prefix]) self.assertIn('put', schema['paths'][prefix]) self.assertIn(f'{prefix}/{{{resource}_ids}}', schema['paths']) self.assertIn('delete', schema['paths'][f'{prefix}/{{{resource}_ids}}']) self.assertIn(f'{prefix}/{{{resource}_id}}', schema['paths']) self.assertIn('get', schema['paths'][f'{prefix}/{{{resource}_id}}']) def test_legacy_delete_openapi_declares_fixed_rejection(self) -> None: app = FastAPI() for router in self.ROUTERS.values(): app.include_router(router) schema = app.openapi() expected_summary = '兼容旧接口，固定拒绝，请使用测试数据批次清理' expected_description = '兼容旧接口，固定拒绝单条物理删除，请使用测试数据批次清理' for resource in self.ROUTERS: with self.subTest(resource=resource): operation = schema['paths'][f'/master/{resource}/{{{resource}_ids}}']['delete'] self.assertEqual(expected_summary, operation['summary']) self.assertEqual(expected_description, operation['description']) self.assertNotIn('用于删除', operation['description']) def test_all_twenty_five_permission_dependencies_are_exact(self) -> None: expected_permissions = { (f'/master/{resource}/list', 'GET'): f'master:{resource}:list' for resource in self.ROUTERS } for resource in self.ROUTERS: expected_permissions.update( { (f'/master/{resource}/{{{resource}_id}}', 'GET'): f'master:{resource}:query', (f'/master/{resource}', 'POST'): f'master:{resource}:add', (f'/master/{resource}', 'PUT'): f'master:{resource}:edit', (f'/master/{resource}/{{{resource}_ids}}', 'DELETE'): f'master:{resource}:list', } ) actual_permissions = {} for router in self.ROUTERS.values(): for route in router.routes: if not isinstance(route, APIRoute): continue method = next(iter(route.methods)) auth_dependencies = [ dependency.dependency for dependency in route.dependencies if isinstance(dependency.dependency, CheckUserInterfaceAuth) ] self.assertEqual(1, len(auth_dependencies)) self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) actual_permissions[(route.path, method)] = auth_dependencies[0].perm self.assertEqual(expected_permissions, actual_permissions) self.assertEqual(25, len(actual_permissions)) def test_write_routes_have_log_and_current_user_dependency(self) -> None: for router in self.ROUTERS.values(): for route in router.routes: if not isinstance(route, APIRoute) or route.methods.isdisjoint({'POST', 'PUT', 'DELETE'}): continue self.assertTrue(hasattr(route.endpoint, '__wrapped__')) self.assertTrue( any( dependency.name == 'current_user' and dependency.call == LoginService.get_current_user for dependency in route.dependant.dependencies ) ) def test_controller_source_has_three_log_and_current_user_declarations(self) -> None: for resource, module in self.MODULES.items(): with self.subTest(resource=resource): source = inspect.getsource(module) self.assertEqual(3, source.count('@Log(')) self.assertEqual(3, source.count('CurrentUserDependency()')) self.assertNotIn('ApiCache', source) def test_route_response_model_contracts(self) -> None: model_by_resource = { 'strain': StrainModel, 'customer': CustomerModel, 'site': SiteModel, 'line': LineModel, 'warehouse': WarehouseModel, } for resource, router in self.ROUTERS.items(): for route in router.routes: if not isinstance(route, APIRoute): continue if route.path.endswith('/list'): self.assertEqual(PageResponseModel[model_by_resource[resource]], route.response_model) elif route.methods == {'GET'}: self.assertEqual(DataResponseModel[model_by_resource[resource]], route.response_model) else: self.assertEqual('ResponseBaseModel', route.response_model.__name__) def test_get_and_write_routes_use_read_and_write_batch_scope_dependencies(self) -> None: for router in self.ROUTERS.values(): for route in router.routes: if not isinstance(route, APIRoute): continue method = next(iter(route.methods)) resolvers = [ dependency.call for dependency in route.dependant.dependencies if isinstance(dependency.call, TestBatchScopeResolver) ] with self.subTest(path=route.path, method=method): if method == 'DELETE': self.assertEqual([], resolvers) else: self.assertEqual(1, len(resolvers)) self.assertIs(method in {'POST', 'PUT'}, resolvers[0].write) class TestDeleteModelCoverage(unittest.TestCase): def test_all_delete_models_are_used_by_service_contract(self) -> None: models = ( DeleteStrainModel(strainIds=[1]), DeleteCustomerModel(customerIds=[1]), DeleteSiteModel(siteIds=[1]), DeleteLineModel(lineIds=[1]), DeleteWarehouseModel(warehouseIds=[1]), ) self.assertEqual(5, len(models)) self.assertTrue(all(next(iter(model.model_dump().values())) == [1] for model in models)) if __name__ == '__main__': unittest.main()