161 lines
7.3 KiB
Python
161 lines
7.3 KiB
Python
import io
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from flask import jsonify
|
|
|
|
from app import app
|
|
from blueprints import admin_api as admin_module
|
|
|
|
|
|
class DedupeTotalDataAdminTest(unittest.TestCase):
|
|
def setUp(self):
|
|
app.config.update(TESTING=True, SECRET_KEY='test')
|
|
self.client = app.test_client()
|
|
|
|
def _login(self, user_id=23):
|
|
with self.client.session_transaction() as session:
|
|
session['user_id'] = user_id
|
|
|
|
def test_list_forwards_session_operator_and_username_search(self):
|
|
self._login()
|
|
java_response = {
|
|
'success': True,
|
|
'data': {
|
|
'items': [{
|
|
'id': 91,
|
|
'dataValue': 'B012345678',
|
|
'uploaderUserId': 23,
|
|
'username': 'member-a',
|
|
'createdAt': '2026-07-20T12:30:00',
|
|
}],
|
|
'total': 1,
|
|
'page': 1,
|
|
'pageSize': 15,
|
|
},
|
|
}
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_dedupe_total_data_access',
|
|
return_value=('normal', {'id': 23, 'username': 'member-a'}, None)), \
|
|
patch.object(admin_module, '_proxy_backend_java',
|
|
return_value=(java_response, None, 200)) as proxy:
|
|
response = self.client.get(
|
|
'/api/admin/dedupe-total-data?keyword=B01&username=member&page=1&page_size=15')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
payload = response.get_json()
|
|
self.assertTrue(payload['success'])
|
|
self.assertEqual('member-a', payload['items'][0]['username'])
|
|
self.assertEqual(23, payload['items'][0]['uploader_user_id'])
|
|
self.assertEqual({
|
|
'page': 1,
|
|
'pageSize': 15,
|
|
'keyword': 'B01',
|
|
'username': 'member',
|
|
'operatorId': 23,
|
|
}, proxy.call_args.kwargs['params'])
|
|
|
|
def test_import_binds_session_operator_to_java_request(self):
|
|
self._login(31)
|
|
java_response = {'success': True, 'message': '开始导入', 'data': {'importId': 'import-1'}}
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_dedupe_total_data_access',
|
|
return_value=('normal', {'id': 31, 'username': 'member-b'}, None)), \
|
|
patch.object(admin_module, '_proxy_backend_java',
|
|
return_value=(java_response, None, 200)) as proxy:
|
|
response = self.client.post(
|
|
'/api/admin/dedupe-total-data/import',
|
|
data={'file': (io.BytesIO(b'excel'), 'data.xlsx')},
|
|
content_type='multipart/form-data')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual('import-1', response.get_json()['import_id'])
|
|
self.assertEqual({'operatorId': 31}, proxy.call_args.kwargs['data'])
|
|
|
|
def test_export_forwards_filters_and_session_operator(self):
|
|
self._login(31)
|
|
java_response = Mock(
|
|
status_code=200,
|
|
content=b'xlsx-data',
|
|
headers={
|
|
'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
|
'Content-Disposition': 'attachment; filename="dedupe-total-data.xlsx"',
|
|
},
|
|
)
|
|
java_session = Mock()
|
|
java_session.get.return_value = java_response
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_dedupe_total_data_access',
|
|
return_value=('normal', {'id': 31, 'username': 'member-b'}, None)), \
|
|
patch.object(admin_module, '_get_backend_java_session', return_value=java_session):
|
|
response = self.client.get(
|
|
'/api/admin/dedupe-total-data/export'
|
|
'?username=member&start_date=2026-07-01&end_date=2026-07-20')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual(b'xlsx-data', response.data)
|
|
self.assertEqual({
|
|
'operatorId': 31,
|
|
'username': 'member',
|
|
'startDate': '2026-07-01',
|
|
'endDate': '2026-07-20',
|
|
}, java_session.get.call_args.kwargs['params'])
|
|
self.assertEqual(60, java_session.get.call_args.kwargs['timeout'])
|
|
|
|
def test_delete_ignores_client_operator_and_uses_session_user(self):
|
|
self._login(41)
|
|
java_response = {'success': True, 'message': '删除成功', 'data': None}
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_dedupe_total_data_access',
|
|
return_value=('admin', {'id': 41, 'username': 'leader'}, None)), \
|
|
patch.object(admin_module, '_proxy_backend_java',
|
|
return_value=(java_response, None, 200)) as proxy:
|
|
response = self.client.delete('/api/admin/dedupe-total-data/91?operatorId=1')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual({'operatorId': 41}, proxy.call_args.kwargs['params'])
|
|
|
|
def test_menu_permission_is_still_required_for_normal_user(self):
|
|
self._login()
|
|
|
|
def deny_access():
|
|
return 'normal', {'id': 23}, (jsonify({'success': False, 'error': '无权访问'}), 403)
|
|
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_dedupe_total_data_access', side_effect=deny_access), \
|
|
patch.object(admin_module, '_proxy_backend_java') as proxy:
|
|
response = self.client.get('/api/admin/dedupe-total-data')
|
|
|
|
self.assertEqual(403, response.status_code)
|
|
self.assertFalse(response.get_json()['success'])
|
|
proxy.assert_not_called()
|
|
|
|
def test_dedupe_access_uses_login_capable_backend_menu_check(self):
|
|
expected = ('normal', {'id': 23}, None)
|
|
with patch.object(admin_module, '_ensure_backend_menu_access', return_value=expected) as backend_access, \
|
|
patch.object(admin_module, '_ensure_admin_menu_access') as admin_access:
|
|
result = admin_module._ensure_dedupe_total_data_access()
|
|
|
|
self.assertEqual(expected, result)
|
|
backend_access.assert_called_once_with('dedupe-total-data')
|
|
admin_access.assert_not_called()
|
|
|
|
def test_group_list_accepts_dedupe_menu_as_access_source(self):
|
|
self._login()
|
|
access_result = ('normal', {'id': 23, 'username': 'member-a'}, None)
|
|
with patch('utils.auth.is_session_user_valid', return_value=True), \
|
|
patch.object(admin_module, '_ensure_backend_menu_access',
|
|
return_value=access_result) as backend_access, \
|
|
patch.object(admin_module, '_load_expanded_shop_manage_groups',
|
|
return_value=([], None, 200)):
|
|
response = self.client.get('/api/admin/shop-manage-groups')
|
|
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertTrue(response.get_json()['success'])
|
|
backend_access.assert_called_once_with(
|
|
'shop-manage', 'skip-price-asin', 'query-asin', 'dedupe-total-data')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|