206 lines
11 KiB
Python
206 lines
11 KiB
Python
"""Regression coverage for the spot authorization-materials workflow."""
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
from pathlib import Path
|
|
from werkzeug.datastructures import MultiDict
|
|
|
|
from app import app
|
|
from routes import spot
|
|
|
|
|
|
class SpotAuthorizationTest(unittest.TestCase):
|
|
def edit_request(self, data, existing=None, failure=False):
|
|
conn = MagicMock()
|
|
cur = conn.cursor.return_value.__enter__.return_value
|
|
cur.fetchone.return_value = existing
|
|
if failure:
|
|
def execute(sql, params):
|
|
if sql.startswith('UPDATE'):
|
|
raise RuntimeError('simulated write failure')
|
|
cur.execute.side_effect = execute
|
|
temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temp.cleanup)
|
|
with app.test_request_context('/api/spot/authorization/23/update', method='POST', data=data), \
|
|
patch.object(spot, 'AUTHORIZATION_UPLOAD_FOLDER', temp.name), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'operation_required', return_value=True) as permission, \
|
|
patch.object(spot, 'get_db', return_value=conn), \
|
|
patch.object(spot, 'add_operation_log'):
|
|
result = spot.authorization_create.__wrapped__(23)
|
|
return result, conn, cur, Path(temp.name), permission
|
|
|
|
def test_edit_retains_existing_attachments_and_updates_text(self):
|
|
result, conn, cur, _, permission = self.edit_request(
|
|
{'merchant_code':'NEW','product_name':'修改商品','remark':'新备注'},
|
|
{'authorization_letter':'old.png','testing_report':'report.pdf'})
|
|
self.assertEqual(result.get_json()['code'], 0)
|
|
values = cur.execute.call_args.args[1]
|
|
self.assertEqual(values[:2], ('NEW','修改商品'))
|
|
self.assertEqual(values[4:6], ('old.png','report.pdf'))
|
|
self.assertEqual(values[8:], ('新备注',23))
|
|
permission.assert_called_once_with('spot_authorization','edit')
|
|
conn.commit.assert_called_once()
|
|
|
|
def test_edit_replaces_uploaded_attachment_and_removes_selected_one(self):
|
|
result, _, cur, folder, _ = self.edit_request(
|
|
{'merchant_code':'M','product_name':'商品','remove_testing_report':'1',
|
|
'authorization_letter':(io.BytesIO(b'new-image'),'new.png')},
|
|
{'authorization_letter':'old.png','testing_report':'old.pdf'})
|
|
self.assertEqual(result.get_json()['code'], 0)
|
|
values=cur.execute.call_args.args[1]
|
|
self.assertNotEqual(values[4], 'old.png')
|
|
self.assertEqual((folder/values[4]).read_bytes(),b'new-image')
|
|
self.assertEqual(values[5], '')
|
|
|
|
def test_missing_record_cleans_new_upload_without_inserting(self):
|
|
result, conn, cur, folder, _ = self.edit_request(
|
|
{'merchant_code':'M','product_name':'商品','authorization_letter':(io.BytesIO(b'image'),'new.png')})
|
|
self.assertEqual(result[1],404)
|
|
self.assertEqual(list(folder.iterdir()),[])
|
|
self.assertEqual(cur.execute.call_count,1)
|
|
conn.commit.assert_not_called()
|
|
|
|
def test_edit_failure_rolls_back_and_cleans_new_upload(self):
|
|
result, conn, _, folder, _ = self.edit_request(
|
|
{'merchant_code':'M','product_name':'商品','authorization_letter':(io.BytesIO(b'image'),'new.png')},
|
|
{'id':23},failure=True)
|
|
self.assertEqual(result.get_json()['code'],500)
|
|
conn.rollback.assert_called_once()
|
|
conn.commit.assert_not_called()
|
|
self.assertEqual(list(folder.iterdir()),[])
|
|
|
|
def test_edit_and_delete_enforce_permissions_before_db(self):
|
|
for action in ('edit','delete'):
|
|
with app.test_request_context('/',method='POST'), \
|
|
patch.object(spot,'role_required',return_value=True), \
|
|
patch.object(spot,'operation_required',return_value=False), \
|
|
patch.object(spot,'get_db') as db:
|
|
fn=spot.authorization_create if action=='edit' else spot.authorization_delete
|
|
self.assertEqual(fn.__wrapped__(23)[1],403)
|
|
db.assert_not_called()
|
|
|
|
def test_delete_targets_record_and_preserves_uploaded_files(self):
|
|
conn=MagicMock()
|
|
cur=conn.cursor.return_value.__enter__.return_value
|
|
cur.fetchone.return_value={'id':23,'merchant_code':'M','product_name':'商品'}
|
|
with app.test_request_context('/',method='POST'), \
|
|
patch.object(spot,'role_required',return_value=True), \
|
|
patch.object(spot,'operation_required',return_value=True), \
|
|
patch.object(spot,'get_db',return_value=conn), \
|
|
patch.object(spot,'add_operation_log'), patch.object(spot.os,'unlink') as unlink:
|
|
result=spot.authorization_delete.__wrapped__(23)
|
|
self.assertEqual(result.get_json()['code'],0)
|
|
self.assertEqual(cur.execute.call_args.args,('DELETE FROM spot_authorization_materials WHERE id=%s',(23,)))
|
|
conn.commit.assert_called_once()
|
|
unlink.assert_not_called()
|
|
|
|
def test_create_persists_product_fields_and_optional_attachments(self):
|
|
with app.test_request_context(
|
|
'/api/spot/authorization/create',
|
|
method='POST',
|
|
data={
|
|
'merchant_code': 'M-001',
|
|
'product_name': '测试商品',
|
|
'platform': '京东',
|
|
'operator': '运营甲',
|
|
'remark': '授权至年底',
|
|
'authorization_letter': (io.BytesIO(b'fake-image'), 'letter.png'),
|
|
},
|
|
) as ctx:
|
|
from flask import session
|
|
session['user_id'] = 17
|
|
temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temp.cleanup)
|
|
with patch.object(spot, 'AUTHORIZATION_UPLOAD_FOLDER', temp.name), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'operation_required', return_value=True), \
|
|
patch.object(spot, 'query', return_value=23) as query, \
|
|
patch.object(spot, 'add_operation_log'):
|
|
response = spot.authorization_create.__wrapped__().get_json()
|
|
self.assertEqual(response['code'], 0)
|
|
sql, params = query.call_args.args
|
|
self.assertIn('INSERT INTO spot_authorization_materials', sql)
|
|
self.assertEqual(params[:5], ('M-001', '测试商品', '京东', '运营甲', params[4]))
|
|
self.assertTrue(params[4].endswith('.png'))
|
|
self.assertEqual(params[8:], ('授权至年底', 17))
|
|
|
|
def test_create_requires_merchant_code_and_product_name(self):
|
|
with app.test_request_context('/api/spot/authorization/create', method='POST', data={}):
|
|
with patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'operation_required', return_value=True), \
|
|
patch.object(spot, 'query') as query:
|
|
response = spot.authorization_create.__wrapped__().get_json()
|
|
self.assertEqual(response['code'], 400)
|
|
query.assert_not_called()
|
|
|
|
def test_create_accepts_two_images_for_multi_document_fields(self):
|
|
data = [
|
|
('merchant_code', 'M-002'), ('product_name', '双证商品'),
|
|
('factory_qualification', (io.BytesIO(b'license'), 'license.png')),
|
|
('factory_qualification', (io.BytesIO(b'permit'), 'permit.jpg')),
|
|
('trademark', (io.BytesIO(b'trade'), 'trade.png')),
|
|
('trademark', (io.BytesIO(b'transfer'), 'transfer.webp')),
|
|
]
|
|
with app.test_request_context('/api/spot/authorization/create', method='POST', data=MultiDict(data)):
|
|
from flask import session
|
|
session['user_id'] = 17
|
|
temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temp.cleanup)
|
|
with patch.object(spot, 'AUTHORIZATION_UPLOAD_FOLDER', temp.name), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'operation_required', return_value=True), \
|
|
patch.object(spot, 'query', return_value=24) as query, \
|
|
patch.object(spot, 'add_operation_log'):
|
|
response = spot.authorization_create.__wrapped__().get_json()
|
|
self.assertEqual(response['code'], 0)
|
|
params = query.call_args.args[1]
|
|
import json
|
|
self.assertEqual(len(json.loads(params[6])), 2)
|
|
self.assertEqual(len(json.loads(params[7])), 2)
|
|
|
|
def test_create_rejects_file_count_over_field_limit(self):
|
|
data = [
|
|
('merchant_code', 'M-003'), ('product_name', '超量商品'),
|
|
('authorization_letter', (io.BytesIO(b'a'), 'letter.png')),
|
|
('authorization_letter', (io.BytesIO(b'b'), 'letter2.png')),
|
|
]
|
|
with app.test_request_context('/api/spot/authorization/create', method='POST', data=MultiDict(data)), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'operation_required', return_value=True), \
|
|
patch.object(spot, 'query') as query:
|
|
response = spot.authorization_create.__wrapped__().get_json()
|
|
self.assertEqual(response['code'], 400)
|
|
self.assertIn('最多上传1个', response['msg'])
|
|
query.assert_not_called()
|
|
|
|
def test_testing_report_rejects_images(self):
|
|
data = MultiDict([
|
|
('merchant_code', 'M-004'), ('product_name', '检测资料'),
|
|
('testing_report', (io.BytesIO(b'image'), 'report.png')),
|
|
])
|
|
with app.test_request_context('/api/spot/authorization/create', method='POST', data=data), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'operation_required', return_value=True), \
|
|
patch.object(spot, 'query') as query:
|
|
response = spot.authorization_create.__wrapped__().get_json()
|
|
self.assertEqual(response['code'], 400)
|
|
self.assertIn('PDF、Word或Excel', response['msg'])
|
|
query.assert_not_called()
|
|
|
|
def test_authorization_page_frontend_contains_requested_columns_and_uploads(self):
|
|
from pathlib import Path
|
|
source = (Path(__file__).resolve().parents[1] / 'static/js/app.js').read_text()
|
|
for label in ('序号', '商家编码', '货品名称', '平台', '负责运营', '授权书', '检测报告', '工厂资质', '商标', '备注'):
|
|
self.assertIn(label, source)
|
|
self.assertIn("openSpotAuthorizationModal", source)
|
|
self.assertIn("picker('testing_report'", source)
|
|
self.assertIn('.spot-auth-modal form{display:flex;flex:1 1 auto;flex-direction:column;min-height:0;overflow:hidden}', source)
|
|
self.assertIn('.spot-auth-modal-foot{display:flex;flex:0 0 auto;', source)
|
|
self.assertIn('button type="submit" class="btn btn-primary">保存资料</button>', source)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|