crawler-plugin/backend/tests/test_admin_dedupe_total_data.py

import io
import sys
import unittest
from pathlib import Path
from unittest.mock import patch

from flask import Flask

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from blueprints import admin_api


class _FakeExportResponse:
    status_code = 200
    content = b'xlsx'
    headers = {
        'Content-Disposition': 'attachment; filename=dedupe.xlsx',
        'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
    }

    def iter_content(self, chunk_size=None):
        yield self.content

    def close(self):
        return None


class AdminDedupeTotalDataTest(unittest.TestCase):
    def setUp(self):
        self.app = Flask(__name__)

    def _menu_access(self):
        return patch.object(
            admin_api,
            '_ensure_backend_menu_access',
            return_value=('admin', {'id': 7}, None),
        )

    def test_list_forwards_group_and_formats_item(self):
        java_response = {
            'data': {
                'items': [{
                    'id': 9,
                    'dataValue': 'B012345678',
                    'groupId': 3,
                    'groupName': 'group-a',
                    'uploaderUserId': 7,
                    'username': 'operator',
                    'createdAt': '2026-08-08T12:30:00',
                }],
                'total': 1,
                'page': 1,
                'pageSize': 15,
            },
        }
        with self.app.test_request_context(
                '/api/admin/dedupe-total-data?page=1&page_size=15&group_id=3'):
            with self._menu_access(), patch.object(
                admin_api,
                '_proxy_permission_java',
                return_value=(java_response, None, 200),
            ) as proxy:
                response = admin_api.list_dedupe_total_data.__wrapped__()

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.get_json()['items'][0]['group_name'], 'group-a')
        self.assertEqual(proxy.call_args.kwargs['params'], {
            'page': 1,
            'pageSize': 15,
            'keyword': '',
            'username': '',
            'operatorId': 7,
            'groupId': 3,
        })

    def test_create_and_update_forward_group(self):
        java_response = {
            'message': 'ok',
            'data': {'id': 9, 'dataValue': 'B012345678', 'groupId': 3, 'groupName': 'group-a'},
        }
        with self.app.test_request_context(
                '/api/admin/dedupe-total-data', method='POST',
                json={'data_value': ' B012345678 ', 'group_id': 3}):
            with self._menu_access(), patch.object(
                admin_api,
                '_proxy_permission_java',
                return_value=(java_response, None, 200),
            ) as proxy:
                response = admin_api.create_dedupe_total_data.__wrapped__()

        self.assertEqual(response.status_code, 200)
        self.assertEqual(proxy.call_args.kwargs['json_data'], {
            'dataValue': 'B012345678',
            'groupId': 3,
        })

        with self.app.test_request_context(
                '/api/admin/dedupe-total-data/9', method='PUT',
                json={'data_value': ' C012345678 ', 'group_id': 8}):
            with self._menu_access(), patch.object(
                admin_api,
                '_proxy_permission_java',
                return_value=(java_response, None, 200),
            ) as proxy:
                response = admin_api.update_dedupe_total_data.__wrapped__(9)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(proxy.call_args.kwargs['json_data'], {
            'dataValue': 'C012345678',
            'groupId': 8,
        })

    def test_import_forwards_multipart_group(self):
        with self.app.test_request_context(
                '/api/admin/dedupe-total-data/import', method='POST',
                data={'file': (io.BytesIO(b'xlsx'), 'data.xlsx'), 'group_id': '3'}):
            with self._menu_access(), patch.object(
                admin_api,
                '_proxy_permission_java',
                return_value=({'message': 'started', 'data': {'importId': 'task-1'}}, None, 200),
            ) as proxy:
                response = admin_api.import_dedupe_total_data.__wrapped__()

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.get_json()['import_id'], 'task-1')
        self.assertEqual(proxy.call_args.kwargs['data'], {'groupId': 3})
        self.assertEqual(set(proxy.call_args.kwargs['files']), {'file'})

    def test_export_forwards_group_filter_and_internal_token(self):
        class _Session:
            def get(self, *args, **kwargs):
                self.args = args
                self.kwargs = kwargs
                return _FakeExportResponse()

        session = _Session()
        with self.app.test_request_context(
                '/api/admin/dedupe-total-data/export?group_id=3&username=operator'):
            with self._menu_access(), patch.object(
                admin_api,
                '_get_backend_java_session',
                return_value=session,
            ), patch.object(admin_api, '_resolve_internal_token', return_value='token'):
                response = admin_api.export_dedupe_total_data.__wrapped__()

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.get_data(), b'xlsx')
        self.assertEqual(session.kwargs['params'], {
            'operatorId': 7,
            'username': 'operator',
            'groupId': 3,
        })
        self.assertEqual(session.kwargs['headers'], {'X-Internal-Token': 'token'})
        self.assertTrue(session.kwargs['stream'])
        self.assertEqual(session.kwargs['timeout'], (10, 1800))

    def test_import_requires_group(self):
        with self.app.test_request_context(
                '/api/admin/dedupe-total-data/import', method='POST',
                data={'file': (io.BytesIO(b'xlsx'), 'data.xlsx')}):
            with self._menu_access(), patch.object(admin_api, '_proxy_permission_java') as proxy:
                response = admin_api.import_dedupe_total_data.__wrapped__()

        self.assertEqual(response.status_code, 200)
        self.assertFalse(response.get_json()['success'])
        proxy.assert_not_called()


if __name__ == '__main__':
    unittest.main()