更新ASIN库新需求

backend/tests/test_dedupe_total_data_admin.py

@@ -0,0 +1,130 @@
+import io
+import unittest
+from unittest.mock import 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_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()