"""新增申请校验与并发查重回归测试；数据库和操作日志全部模拟。""" import unittest from unittest.mock import MagicMock, patch from flask import Flask, session from routes import spot class ApplicationValidationTest(unittest.TestCase): def setUp(self): self.app = Flask(__name__) self.app.secret_key = 'test-only' self.data = dict(product_name='样品', barcode='0012345678901', sample_quantity='2', recipient_info='收件人及地址', group_name='京东1组') self.conn = MagicMock() self.cur = self.conn.cursor.return_value.__enter__.return_value self.cur.fetchone.side_effect = [{'acquired': 1}, None] self.cur.lastrowid = 88 self.db = patch.object(spot, 'get_db', return_value=self.conn).start() self.log = patch.object(spot, 'add_operation_log').start() patch.object(spot, 'role_required', return_value=True).start() self.addCleanup(patch.stopall) def create(self, changes=None): data = dict(self.data) data.update(changes or {}) with self.app.test_request_context('/api/spot/application/create', method='POST', data=data): session['user_id'] = 7 return spot.application_create.__wrapped__().get_json() def test_all_required_fields_are_checked_before_db_access(self): for key in ('product_name', 'barcode', 'recipient_info', 'sample_quantity'): for value in ('', ' '): with self.subTest(key=key, value=value): self.assertEqual(self.create({key: value})['code'], 400) self.db.assert_not_called() def test_barcode_requires_exactly_thirteen_ascii_digits(self): for value in ('123456789012', '12345678901234', 'a123456789012', '１２３４５６７８９０１２３', '123456 7890123', '1e12345678901'): with self.subTest(value=value): response = self.create({'barcode': value}) self.assertEqual(response['code'], 400) self.assertIn('请填写正确的条形码', response['msg']) self.db.assert_not_called() def test_valid_group_is_saved_and_leading_zeroes_are_preserved(self): response = self.create() self.assertEqual(response, {'code': 0, 'id': 88, 'msg': '申请已添加'}) insert = next(c for c in self.cur.execute.call_args_list if 'INSERT INTO spot_applications' in c.args[0]) self.assertIn('group_name', insert.args[0]) self.assertEqual(insert.args[1][2], '京东1组') self.assertEqual(insert.args[1][-2], '0012345678901') self.conn.commit.assert_called_once() self.assertIn('RELEASE_LOCK', self.cur.execute.call_args.args[0]) self.conn.close.assert_called_once() def test_each_allowed_group_and_blank_are_accepted(self): for group in (*spot.SPOT_APPLICATION_GROUPS, ''): self.cur.fetchone.side_effect = [{'acquired': 1}, None] with self.subTest(group=group): self.assertEqual(self.create({'group_name': group})['code'], 0) def test_invalid_group_is_rejected(self): self.assertEqual(self.create({'group_name': '其他组'})['code'], 400) self.db.assert_not_called() def test_department_options_are_saved(self): for department in ('洁维塔', '泊紫汀兰', ''): self.cur.fetchone.side_effect = [{'acquired': 1}, None] self.assertEqual(self.create({'department': department})['code'], 0) insert = next(c for c in reversed(self.cur.execute.call_args_list) if 'INSERT INTO' in c.args[0]) self.assertEqual(insert.args[1][1], department) def test_invalid_department_is_rejected(self): for changes in ({'department': '其他部门'},): self.assertEqual(self.create(changes)['code'], 400) self.db.assert_not_called() def test_jieweita_requires_group_before_database_access(self): for group in ('', ' '): result = self.create({'brand': '洁维塔', 'group_name': group}) self.assertEqual(result['code'], 400) self.assertIn('必须选择组别', result['msg']) self.db.assert_not_called() def test_bozitinglan_accepts_blank_group(self): self.assertEqual(self.create({'brand': '泊紫汀兰', 'group_name': ''})['code'], 0) def test_shenzhen_template_overrides_tampered_recipient_fields(self): result = self.create(dict(brand='泊紫汀兰', recipient_template='shenzhen', recipient_address='其他地址', recipient_name='其他人', recipient_phone='000', recipient_info='篡改的信息')) self.assertEqual(result['code'], 0) insert = next(c for c in self.cur.execute.call_args_list if 'INSERT INTO' in c.args[0]) self.assertEqual(insert.args[1][-1], '深圳市龙岗区坂田街道发达路6号云里智能园7栋3楼蝶雅东侧\n收件人：一部样品\n电话：17771977816') def test_guangzhou_template_accepts_contact_edits_but_keeps_address(self): result = self.create(dict(brand='洁维塔', recipient_template='guangzhou', recipient_address='其他地址', recipient_name=' 张三 ', recipient_phone=' 13800138000 ')) self.assertEqual(result['code'], 0) insert = next(c for c in self.cur.execute.call_args_list if 'INSERT INTO' in c.args[0]) self.assertEqual(insert.args[1][-1], '广州市白云区鹤龙街道广云路238号思成未来城市总部5楼\n收件人：张三\n电话：13800138000') def test_invalid_template_and_empty_guangzhou_contacts_are_rejected(self): for changes in ({'recipient_template': ''}, {'recipient_template': 'other'}, {'recipient_template': 'guangzhou', 'recipient_name': ' ', 'recipient_phone': '123'}, {'recipient_template': 'guangzhou', 'recipient_name': '张三', 'recipient_phone': ' '}): self.assertEqual(self.create(changes)['code'], 400) self.db.assert_not_called() def test_free_text_brand_is_accepted(self): self.assertEqual(self.create({'brand': ' 其他品牌 '})['code'], 0) insert = next(c for c in self.cur.execute.call_args_list if 'INSERT INTO' in c.args[0]) self.assertIn('其他品牌', insert.args[1]) def test_overlong_brand_is_rejected(self): self.assertEqual(self.create({'brand': '品' * 101})['code'], 400) self.db.assert_not_called() def test_duplicate_name_or_barcode_is_rejected_under_lock(self): self.cur.fetchone.side_effect = [{'acquired': 1}, {'id': 3}] response = self.create() self.assertEqual(response['code'], 409) self.assertIn('存在重复值', response['msg']) calls = self.cur.execute.call_args_list self.assertIn('GET_LOCK', calls[0].args[0]) self.assertIn('TRIM(product_name)=%s OR TRIM(barcode)=%s', calls[1].args[0]) self.assertEqual(calls[1].args[1], ('样品', '0012345678901')) self.assertFalse(any('INSERT INTO' in c.args[0] for c in calls)) self.conn.commit.assert_not_called() self.assertIn('RELEASE_LOCK', calls[-1].args[0]) def test_lock_timeout_never_inserts(self): self.cur.fetchone.side_effect = [{'acquired': 0}] self.assertEqual(self.create()['code'], 409) self.conn.commit.assert_not_called() self.assertFalse(any('INSERT INTO' in c.args[0] for c in self.cur.execute.call_args_list)) self.conn.close.assert_called_once() def test_negative_fractional_and_overflow_quantity_rejected(self): for quantity in ('-1', '1.5', '2147483648'): self.assertEqual(self.create({'sample_quantity': quantity})['code'], 400) self.db.assert_not_called() def test_audit_failure_does_not_report_already_saved_request_as_failed(self): self.log.side_effect = RuntimeError('audit unavailable') self.assertEqual(self.create()['code'], 0) self.conn.commit.assert_called_once() def test_insert_failure_rolls_back_and_releases_lock(self): def execute(sql, params=None): if 'INSERT INTO' in sql: raise RuntimeError('database unavailable') self.cur.execute.side_effect = execute self.assertEqual(self.create()['code'], 500) self.conn.commit.assert_not_called() self.conn.rollback.assert_called() self.assertIn('RELEASE_LOCK', self.cur.execute.call_args.args[0]) if __name__ == '__main__': unittest.main()