168 lines
8.6 KiB
Python
168 lines
8.6 KiB
Python
"""新增申请校验与并发查重回归测试；数据库和操作日志全部模拟。"""
|
||
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()
|