340 lines
17 KiB
Python
340 lines
17 KiB
Python
import io
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from exceptions.exception import PermissionException, ServiceException
|
|
from module_business.dao.master_dao import MasterDao
|
|
from module_business.domain.test_batch_scope import TestBatchScope
|
|
from module_business.entity.do.master_do import BizPallet
|
|
from module_business.entity.do.operation_do import BizPalletInventory
|
|
from module_business.entity.vo.master_vo import StrainModel, StrainPageQueryModel
|
|
from module_business.service.master_excel_service import MasterExcelService, _LISTERS
|
|
from module_business.service.master_service import MasterService
|
|
from tests.module_business.test_batch_fixtures import make_admin, make_current_user, make_test_scope
|
|
|
|
|
|
class _Upload:
|
|
def __init__(self, content: bytes) -> None:
|
|
self.content = content
|
|
self.closed = False
|
|
|
|
async def read(self) -> bytes:
|
|
return self.content
|
|
|
|
async def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def workbook_bytes(headers: list[str], rows: list[list[object]]) -> bytes:
|
|
workbook = Workbook()
|
|
worksheet = workbook.active
|
|
worksheet.append(headers)
|
|
for row in rows:
|
|
worksheet.append(row)
|
|
buffer = io.BytesIO()
|
|
workbook.save(buffer)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def sheet_rows(binary: bytes) -> list[tuple[object, ...]]:
|
|
workbook = load_workbook(io.BytesIO(binary), data_only=False)
|
|
try:
|
|
return [tuple(cell.value for cell in row) for row in workbook.active.iter_rows()]
|
|
finally:
|
|
workbook.close()
|
|
|
|
|
|
def compile_sql(statement: object) -> str:
|
|
return ' '.join(str(statement.compile()).lower().split())
|
|
|
|
|
|
class MasterExcelServiceTest(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.db = AsyncMock(spec=AsyncSession)
|
|
self.admin = make_admin()
|
|
self.scope = TestBatchScope.formal()
|
|
self.executed: list[object] = []
|
|
|
|
async def execute(statement: object) -> None:
|
|
self.executed.append(statement)
|
|
|
|
self.db.execute = execute
|
|
|
|
async def _import(self, resource: str, headers: list[str], rows: list[list[object]], *, update_support: bool = False, scope=None, user=None):
|
|
return await MasterExcelService.import_resource(
|
|
self.db,
|
|
resource, # type: ignore[arg-type]
|
|
_Upload(workbook_bytes(headers, rows)),
|
|
update_support,
|
|
user or self.admin,
|
|
scope or self.scope,
|
|
)
|
|
|
|
async def test_add_skip_update_and_other_scope(self) -> None:
|
|
headers = ['菌种编码', '菌种名称', '状态', '备注']
|
|
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(
|
|
MasterDao, 'add_strain', new=AsyncMock()
|
|
) as add_strain:
|
|
added = await self._import('strain', headers, [[1001, '香菇', '正常', '首批']])
|
|
self.assertIn('新增1条', added.message)
|
|
saved = add_strain.await_args.args[1]
|
|
self.assertEqual('1001', saved.strain_code)
|
|
self.assertEqual('香菇', saved.strain_name)
|
|
self.assertEqual('0', saved.status)
|
|
self.assertIsNone(add_strain.await_args.args[2])
|
|
self.db.commit.assert_awaited_once()
|
|
|
|
self.db.commit.reset_mock()
|
|
existing = SimpleNamespace(strain_id=3, strain_code='S001', test_batch_id=None)
|
|
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=existing)), patch.object(
|
|
MasterDao, 'get_strain', new=AsyncMock(return_value=existing)
|
|
), patch.object(MasterDao, 'update_strain', new=AsyncMock()) as update_strain:
|
|
skipped = await self._import('strain', headers, [['S001', '香菇', '正常', '']])
|
|
self.assertIn('跳过1条', skipped.message)
|
|
self.assertIn('已跳过', skipped.message)
|
|
update_strain.assert_not_awaited()
|
|
self.db.rollback.assert_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
|
|
self.db.rollback.reset_mock()
|
|
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=existing)), patch.object(
|
|
MasterDao, 'get_strain', new=AsyncMock(return_value=existing)
|
|
), patch.object(MasterDao, 'update_strain', new=AsyncMock()) as update_strain:
|
|
updated = await self._import('strain', headers, [['S001', '新香菇', '停用', '改备注']], update_support=True)
|
|
self.assertIn('更新1条', updated.message)
|
|
model = update_strain.await_args.args[1]
|
|
self.assertEqual(3, model.strain_id)
|
|
self.assertEqual('S001', model.strain_code)
|
|
self.assertEqual('新香菇', model.strain_name)
|
|
self.assertEqual('1', model.status)
|
|
self.db.commit.assert_awaited_once()
|
|
|
|
self.db.commit.reset_mock()
|
|
outside = SimpleNamespace(strain_id=9, strain_code='S009', test_batch_id=4)
|
|
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=outside)), patch.object(
|
|
MasterDao, 'get_strain', new=AsyncMock(return_value=None)
|
|
), patch.object(MasterDao, 'add_strain', new=AsyncMock()) as add_strain:
|
|
rejected = await self._import('strain', headers, [['S009', '越界', '正常', '']])
|
|
self.assertIn('已存在于其他数据范围', rejected.message)
|
|
add_strain.assert_not_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_awaited()
|
|
|
|
async def test_reordered_headers_and_row_error_still_commit_success_rows(self) -> None:
|
|
headers = ['备注', '状态', '菌种名称', '菌种编码']
|
|
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(
|
|
MasterDao, 'add_strain', new=AsyncMock()
|
|
) as add_strain:
|
|
result = await self._import('strain', headers, [['备注', '正常', '香菇', 'S001'], ['坏', '启用', '平菇', 'S002']])
|
|
self.assertEqual(1, add_strain.await_count)
|
|
self.assertIn('新增1条', result.message)
|
|
self.assertIn('第3行', result.message)
|
|
self.assertIn('状态只能填写正常或停用', result.message)
|
|
self.db.commit.assert_awaited_once()
|
|
|
|
async def test_duplicate_identity_rejects_whole_file_before_write(self) -> None:
|
|
headers = ['菌种编码', '菌种名称', '状态', '备注']
|
|
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock()) as find_strain:
|
|
with self.assertRaises(ServiceException) as caught:
|
|
await self._import('strain', headers, [['S001', '甲', '正常', ''], ['s001', '乙', '停用', '']])
|
|
self.assertIn('重复编码', caught.exception.message)
|
|
find_strain.assert_not_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_not_awaited()
|
|
|
|
async def test_header_mismatch_empty_file_and_size_limit(self) -> None:
|
|
with self.assertRaises(ServiceException) as mismatch:
|
|
await self._import('strain', ['菌种编码', '多余'], [['S001', '甲']])
|
|
self.assertIn('表头不匹配', mismatch.exception.message)
|
|
|
|
with self.assertRaises(ServiceException) as empty:
|
|
await MasterExcelService.import_resource(
|
|
self.db, 'strain', _Upload(b''), False, self.admin, self.scope
|
|
)
|
|
self.assertEqual('导入文件不能为空', empty.exception.message)
|
|
|
|
with self.assertRaises(ServiceException) as oversized:
|
|
await MasterExcelService.import_resource(
|
|
self.db, 'strain', _Upload(b'x' * (5 * 1024 * 1024 + 1)), False, self.admin, self.scope
|
|
)
|
|
self.assertIn('5MB', oversized.exception.message)
|
|
self.db.commit.assert_not_awaited()
|
|
|
|
async def test_missing_parent_and_ambiguous_site_code_are_row_errors(self) -> None:
|
|
site_headers = ['客户编码', '现场编码', '现场名称', '现场地址', '状态', '备注']
|
|
with patch.object(MasterDao, 'find_customer_by_code', new=AsyncMock(return_value=None)), patch.object(
|
|
MasterDao, 'add_site', new=AsyncMock()
|
|
) as add_site:
|
|
missing = await self._import('site', site_headers, [['C404', 'SITE1', '现场', '地址', '正常', '']])
|
|
self.assertIn('目标客户C404不存在', missing.message)
|
|
add_site.assert_not_awaited()
|
|
|
|
line_headers = ['现场编码', '产线编码', '产线名称', '状态', '备注']
|
|
sites = [
|
|
SimpleNamespace(site_id=1, site_code='SITE1', status='0', test_batch_id=None),
|
|
SimpleNamespace(site_id=2, site_code='SITE1', status='0', test_batch_id=None),
|
|
]
|
|
with patch.object(MasterDao, 'list_sites_by_code', new=AsyncMock(return_value=sites)), patch.object(
|
|
MasterDao, 'add_line', new=AsyncMock()
|
|
) as add_line:
|
|
ambiguous = await self._import('line', line_headers, [['SITE1', 'L001', '一线', '正常', '']])
|
|
self.assertIn('现场编码SITE1不唯一', ambiguous.message)
|
|
add_line.assert_not_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
|
|
async def test_database_error_rolls_back_and_permission_error_is_not_wrapped(self) -> None:
|
|
headers = ['菌种编码', '菌种名称', '状态', '备注']
|
|
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(
|
|
MasterDao, 'add_strain', new=AsyncMock(side_effect=IntegrityError('statement', {}, RuntimeError('duplicate')))
|
|
):
|
|
with self.assertRaises(ServiceException) as caught:
|
|
await self._import('strain', headers, [['S001', '香菇', '正常', '']])
|
|
self.assertEqual('导入失败，数据已全部回滚', caught.exception.message)
|
|
self.db.rollback.assert_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
|
|
self.db.rollback.reset_mock()
|
|
with self.assertRaises(PermissionException) as denied:
|
|
await self._import(
|
|
'strain',
|
|
headers,
|
|
[['S001', '香菇', '正常', '']],
|
|
scope=make_test_scope(9),
|
|
user=make_current_user(admin=False),
|
|
)
|
|
self.assertEqual('仅超级管理员可执行测试批次管理操作', denied.exception.message)
|
|
self.db.rollback.assert_awaited()
|
|
|
|
async def test_import_uses_current_test_batch(self) -> None:
|
|
headers = ['菌种编码', '菌种名称', '状态', '备注']
|
|
scope = make_test_scope(9)
|
|
with patch.object(MasterService, '_prepare_test_write', new=AsyncMock()) as prepare, patch.object(
|
|
MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)
|
|
), patch.object(MasterDao, 'add_strain', new=AsyncMock()) as add_strain:
|
|
await self._import('strain', headers, [['S001', '香菇', '正常', '']], scope=scope, user=self.admin)
|
|
prepare.assert_awaited_once()
|
|
self.assertIs(scope, prepare.await_args.args[2])
|
|
self.assertEqual(9, add_strain.await_args.args[2])
|
|
|
|
async def test_new_pallet_gets_fixed_capacity_qr_and_empty_inventory(self) -> None:
|
|
async def flush() -> None:
|
|
for call in self.db.add.call_args_list:
|
|
record = call.args[0]
|
|
if isinstance(record, BizPallet) and record.pallet_id is None:
|
|
record.pallet_id = 15
|
|
|
|
self.db.flush = flush
|
|
with patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)):
|
|
result = await self._import('pallet', ['托盘编号', '状态', '备注'], [['P001', '停用', '新盘']])
|
|
added = [call.args[0] for call in self.db.add.call_args_list]
|
|
pallet = next(record for record in added if isinstance(record, BizPallet))
|
|
inventory = next(record for record in added if isinstance(record, BizPalletInventory))
|
|
self.assertEqual(100, pallet.capacity_packages)
|
|
self.assertEqual('1', pallet.asset_status)
|
|
self.assertTrue(pallet.qr_token)
|
|
self.assertNotEqual('P001', pallet.qr_token)
|
|
self.assertEqual(15, inventory.pallet_id)
|
|
self.assertEqual(0, inventory.remaining_packages)
|
|
self.assertEqual('UNSTOCKED', inventory.flow_state)
|
|
self.assertIsNone(inventory.test_batch_id)
|
|
self.assertIn('新增1条', result.message)
|
|
self.db.commit.assert_awaited_once()
|
|
|
|
async def test_pallet_update_keeps_capacity_and_qr_token(self) -> None:
|
|
pallet = SimpleNamespace(
|
|
pallet_id=5,
|
|
pallet_code='P001',
|
|
qr_token='qr-keep',
|
|
capacity_packages=80,
|
|
asset_status='0',
|
|
remark='旧',
|
|
test_batch_id=None,
|
|
)
|
|
inventory = SimpleNamespace(pallet_id=5, remaining_packages=3, test_batch_id=None)
|
|
with patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=pallet)), patch.object(
|
|
MasterDao, 'get_pallet', new=AsyncMock(return_value=pallet)
|
|
), patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)):
|
|
result = await self._import(
|
|
'pallet', ['托盘编号', '状态', '备注'], [['P001', '停用', '新备注']], update_support=True
|
|
)
|
|
self.assertIn('更新1条', result.message)
|
|
self.assertEqual('qr-keep', pallet.qr_token)
|
|
self.assertEqual(80, pallet.capacity_packages)
|
|
sql = compile_sql(self.executed[-1])
|
|
self.assertNotIn('qr_token', sql)
|
|
self.assertNotIn('pallet_code', sql)
|
|
self.assertIn('capacity_packages', sql)
|
|
self.assertIn('asset_status', sql)
|
|
self.db.commit.assert_awaited_once()
|
|
|
|
async def test_tunnel_update_sql_does_not_touch_current_trace(self) -> None:
|
|
site = SimpleNamespace(site_id=2, site_code='SITE1', status='0', test_batch_id=None)
|
|
tunnel = SimpleNamespace(tunnel_id=8, tunnel_code='T001', site_id=2, status='0', test_batch_id=None)
|
|
with patch.object(MasterDao, 'list_sites_by_code', new=AsyncMock(return_value=[site])), patch.object(
|
|
MasterDao, 'find_tunnel_by_code', new=AsyncMock(return_value=tunnel)
|
|
), patch.object(MasterDao, 'get_tunnel', new=AsyncMock(return_value=tunnel)), patch.object(
|
|
MasterDao, 'get_site', new=AsyncMock(return_value=site)
|
|
):
|
|
result = await self._import(
|
|
'tunnel',
|
|
['现场编码', '隧道编号', '隧道名称', '状态', '备注'],
|
|
[['SITE1', 'T001', '新隧道', '停用', '改备注']],
|
|
update_support=True,
|
|
)
|
|
self.assertIn('更新1条', result.message)
|
|
sql = compile_sql(self.executed[-1])
|
|
self.assertNotIn('current_trace_no', sql)
|
|
self.assertNotIn('current_trace_started_at', sql)
|
|
self.assertIn('tunnel_name', sql)
|
|
self.assertIn('status', sql)
|
|
self.db.commit.assert_awaited_once()
|
|
|
|
async def test_export_uses_chinese_status_and_empty_result_keeps_headers(self) -> None:
|
|
rows = [
|
|
StrainModel(strainId=1, strainCode='S001', strainName='香菇', status='0', remark='甲'),
|
|
StrainModel(strainId=2, strainCode='S002', strainName='平菇', status='1', remark=None),
|
|
]
|
|
query = StrainPageQueryModel()
|
|
list_strain = AsyncMock(return_value=rows)
|
|
with patch.dict(_LISTERS, {'strain': list_strain}):
|
|
binary = await MasterExcelService.export_resource(self.db, 'strain', query, self.scope)
|
|
self.assertFalse(list_strain.await_args.kwargs['is_page'])
|
|
self.assertEqual(
|
|
[
|
|
('菌种编码', '菌种名称', '状态', '备注'),
|
|
('S001', '香菇', '正常', '甲'),
|
|
('S002', '平菇', '停用', None),
|
|
],
|
|
[row[:4] for row in sheet_rows(binary)],
|
|
)
|
|
|
|
with patch.dict(_LISTERS, {'strain': AsyncMock(return_value=[])}):
|
|
empty = await MasterExcelService.export_resource(self.db, 'strain', query, self.scope)
|
|
exported = sheet_rows(empty)
|
|
self.assertEqual([('菌种编码', '菌种名称', '状态', '备注')], [row[:4] for row in exported])
|
|
self.assertEqual(1, len(exported))
|
|
|
|
async def test_export_passes_query_with_ids(self) -> None:
|
|
rows = [
|
|
StrainModel(strainId=1, strainCode='S001', strainName='香菇', status='0', remark='甲'),
|
|
]
|
|
query = StrainPageQueryModel(ids='1')
|
|
list_strain = AsyncMock(return_value=rows)
|
|
with patch.dict(_LISTERS, {'strain': list_strain}):
|
|
binary = await MasterExcelService.export_resource(self.db, 'strain', query, self.scope)
|
|
self.assertFalse(list_strain.await_args.kwargs['is_page'])
|
|
self.assertEqual(query, list_strain.await_args.args[1])
|
|
self.assertEqual('1', list_strain.await_args.args[1].ids)
|
|
self.assertEqual(
|
|
[
|
|
('菌种编码', '菌种名称', '状态', '备注'),
|
|
('S001', '香菇', '正常', '甲'),
|
|
],
|
|
[row[:4] for row in sheet_rows(binary)],
|
|
)
|