1090 lines
48 KiB
Python
1090 lines
48 KiB
Python
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_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}:remove',
|
|
}
|
|
)
|
|
|
|
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()
|