"""采样确认导入校验、权限及事务回归；不写入真实数据库。""" 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()