173 lines
9.4 KiB
Python
173 lines
9.4 KiB
Python
"""采样确认导入校验、权限及事务回归；不写入真实数据库。"""
|
|
import io
|
|
import unittest
|
|
from datetime import date
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from openpyxl import load_workbook
|
|
from app import app
|
|
from services.spot_confirmation_import import (
|
|
ImportValidationError, confirmation_template, import_confirmations,
|
|
parse_confirmation_import,
|
|
)
|
|
|
|
|
|
class ConfirmationImportTest(unittest.TestCase):
|
|
def parse_csv(self, text):
|
|
return parse_confirmation_import(text.encode('utf-8-sig'), 'data.csv')
|
|
|
|
def connection(self, existing=None):
|
|
conn = MagicMock()
|
|
cur = conn.cursor.return_value.__enter__.return_value
|
|
cur.fetchall.return_value = [
|
|
{'id': 1, 'username': 'admin', 'real_name': '管理员'},
|
|
{'id': 2, 'username': 'one', 'real_name': '同名'},
|
|
{'id': 3, 'username': 'two', 'real_name': '同名'},
|
|
]
|
|
cur.fetchone.return_value = existing
|
|
cur.lastrowid = 88
|
|
return conn, cur
|
|
|
|
def test_template_and_excel_roundtrip(self):
|
|
wb = load_workbook(confirmation_template())
|
|
headers = [c.value for c in wb.active[1]]
|
|
self.assertEqual(headers, [
|
|
'申请编号', '日期', '品牌', '产品名称', '图片', '条形码', '采样数量', '产品链接',
|
|
'部门', '组别', '申请人', '厂家名称', '收件信息', '样品状态', '产品单价',
|
|
'OA审批单号', '寄样快递单号', '到样日期', '备注', '领样部门', '领样人',
|
|
'样品确认', '样品确认原因', '产品编码', '货品名称（最终名称）', '平台',
|
|
])
|
|
self.assertIn('填写说明', wb.sheetnames)
|
|
values = {'产品名称': '样品', '条形码': '00123', '采样数量': 0,
|
|
'日期': date(2026, 10, 14)}
|
|
for i, label in enumerate(headers, 1):
|
|
wb.active.cell(2, i, values.get(label))
|
|
buf = io.BytesIO(); wb.save(buf)
|
|
rows = parse_confirmation_import(buf.getvalue(), 'template.xlsx')
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0][1]['barcode'], '00123')
|
|
self.assertEqual(rows[0][1]['sample_quantity'], 0)
|
|
self.assertEqual(rows[0][1]['application_date'], date(2026, 10, 14))
|
|
|
|
def test_empty_template_is_not_data(self):
|
|
with self.assertRaisesRegex(ImportValidationError, '没有可导入'):
|
|
parse_confirmation_import(confirmation_template().getvalue(), 'empty.xlsx')
|
|
|
|
def test_csv_aliases_and_blank_preservation(self):
|
|
rows = self.parse_csv('申请编号,产品名称,寄养数量,厂家\n7,,0,工厂\n')
|
|
self.assertEqual(rows, [(2, {'application_id': 7, 'sample_quantity': 0, 'manufacturer': '工厂'})])
|
|
|
|
def test_invalid_values_and_row_numbers(self):
|
|
for text in ('产品名称,采样数量\n\n品,-1', '产品名称,采样数量\n品,1.2',
|
|
'产品名称,日期\n品,2026-02-30', '产品名称,产品链接\n品,javascript:alert(1)',
|
|
'申请编号\n7\n7', '产品名称,寄样数量,采样数量\n品,1,2',
|
|
'产品名称\n' + '品' * 301):
|
|
with self.subTest(text=text[:30]), self.assertRaises(ImportValidationError):
|
|
self.parse_csv(text)
|
|
with self.assertRaisesRegex(ImportValidationError, '第3行'):
|
|
self.parse_csv('产品名称,采样数量\n\n品,-1')
|
|
|
|
def test_formula_rejected(self):
|
|
wb = load_workbook(confirmation_template()); wb.active['D2'] = '=1+1'
|
|
buf = io.BytesIO(); wb.save(buf)
|
|
with self.assertRaisesRegex(ImportValidationError, '含公式'):
|
|
parse_confirmation_import(buf.getvalue(), 'formula.xlsx')
|
|
|
|
def test_new_record_is_approved_and_links_confirmation(self):
|
|
conn, cur = self.connection()
|
|
result = import_confirmations(self.parse_csv('产品名称,采样数量,样品状态\n样品,2,待寄出'), 1, conn)
|
|
self.assertEqual(result, {'added': 1, 'updated': 0, 'total': 1})
|
|
calls = cur.execute.call_args_list
|
|
insert = next(call for call in calls if call.args[0].startswith('INSERT INTO spot_applications'))
|
|
self.assertIn('通过', insert.args[1])
|
|
self.assertTrue(any(call.args[0].startswith('UPDATE spot_confirmations') and
|
|
call.args[1] == ('待寄出', 88) for call in calls))
|
|
conn.commit.assert_called_once(); conn.rollback.assert_not_called()
|
|
|
|
def test_update_only_supplied_fields(self):
|
|
conn, cur = self.connection({'id': 7, 'status': '通过'})
|
|
result = import_confirmations(self.parse_csv('申请编号,产品名称,采样数量\n7,,0'), 1, conn)
|
|
self.assertEqual(result['updated'], 1)
|
|
update = next(c for c in cur.execute.call_args_list if c.args[0].startswith('UPDATE spot_applications'))
|
|
self.assertNotIn('product_name', update.args[0])
|
|
self.assertEqual(update.args[1], (0, 7))
|
|
|
|
def test_invalid_batch_never_writes(self):
|
|
for existing, text in ((None, '产品名称,申请编号\n新样品,\n,7'),
|
|
({'id': 7, 'status': '待处理'}, '申请编号\n7')):
|
|
conn, cur = self.connection(existing)
|
|
with self.assertRaises(ImportValidationError):
|
|
import_confirmations(self.parse_csv(text), 1, conn)
|
|
self.assertFalse(any(c.args[0].startswith(('INSERT', 'UPDATE')) for c in cur.execute.call_args_list))
|
|
conn.rollback.assert_called_once(); conn.commit.assert_not_called()
|
|
|
|
def test_historical_applicant_imported_as_plain_text(self):
|
|
conn, cur = self.connection()
|
|
rows = self.parse_csv('产品名称,申请人,领样部门,领样人\n历史产品,吴海鹏,京东1组,吴海鹏\n')
|
|
result = import_confirmations(rows, 1, conn)
|
|
self.assertEqual(result['added'], 1)
|
|
app_insert = next(c for c in cur.execute.call_args_list
|
|
if c.args[0].startswith('INSERT INTO spot_applications'))
|
|
self.assertIn('applicant_name', app_insert.args[0])
|
|
self.assertIn('吴海鹏', app_insert.args[1])
|
|
conf_update = next(c for c in cur.execute.call_args_list
|
|
if c.args[0].startswith('UPDATE spot_confirmations'))
|
|
self.assertIn('receiving_department', conf_update.args[0])
|
|
self.assertIn('recipient', conf_update.args[0])
|
|
|
|
def test_department_and_group_are_imported_separately(self):
|
|
conn, cur = self.connection()
|
|
rows = self.parse_csv('产品名称,部门,组别\n样品,洁维塔,京东1组\n')
|
|
import_confirmations(rows, 1, conn)
|
|
app_insert = next(c for c in cur.execute.call_args_list
|
|
if c.args[0].startswith('INSERT INTO spot_applications'))
|
|
self.assertIn('`department`', app_insert.args[0])
|
|
self.assertIn('`group_name`', app_insert.args[0])
|
|
self.assertIn('洁维塔', app_insert.args[1])
|
|
self.assertIn('京东1组', app_insert.args[1])
|
|
|
|
def test_database_error_rolls_back(self):
|
|
conn, cur = self.connection()
|
|
def execute(sql, *args):
|
|
if sql.startswith('UPDATE spot_confirmations'):
|
|
raise RuntimeError('write failed')
|
|
cur.execute.side_effect = execute
|
|
with self.assertRaises(RuntimeError):
|
|
import_confirmations(self.parse_csv('产品名称,备注\n品,说明'), 1, conn)
|
|
conn.rollback.assert_called_once(); conn.commit.assert_not_called()
|
|
|
|
|
|
class ConfirmationRouteTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = app.test_client()
|
|
with self.client.session_transaction() as s:
|
|
s.update(user_id=1, role='super_admin', username='admin')
|
|
|
|
def test_template_download(self):
|
|
response = self.client.get('/api/spot/confirmation/import_template')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('采样数量', [c.value for c in load_workbook(io.BytesIO(response.data)).active[1]])
|
|
|
|
def test_missing_file_and_parse_failure_do_not_connect(self):
|
|
with patch('routes.spot.get_db') as db:
|
|
self.assertEqual(self.client.post('/api/spot/confirmation/import').status_code, 400)
|
|
response = self.client.post('/api/spot/confirmation/import', data={'file': (io.BytesIO('产品名称,采样数量\n品,-1'.encode()), 'bad.csv')})
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn('第2行', response.get_json()['errors'][0]); db.assert_not_called()
|
|
|
|
def test_permission_enforced(self):
|
|
with patch('routes.spot.role_required', return_value=True), patch('routes.spot.operation_required', return_value=False), patch('routes.spot.get_db') as db:
|
|
self.assertEqual(self.client.post('/api/spot/confirmation/import').status_code, 403)
|
|
db.assert_not_called()
|
|
|
|
def test_success_and_postcommit_failure_still_success(self):
|
|
with patch('routes.spot.get_db') as db, patch('routes.spot.import_confirmations', return_value={'added': 1, 'updated': 0, 'total': 1}), patch('routes.spot.bump_data_version', side_effect=RuntimeError('cache unavailable')), patch('routes.spot.add_operation_log') as audit:
|
|
response = self.client.post('/api/spot/confirmation/import', data={'file': (io.BytesIO('产品名称\n品'.encode()), 'data.csv')})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.get_json()['code'], 0)
|
|
db.return_value.close.assert_called_once(); audit.assert_called_once()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|