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()