supernijia
200 lines
8.1 KiB
Python
200 lines
8.1 KiB
Python
import sys
|
|
import unittest
|
|
from datetime import datetime
|
|
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 _FakeCursor:
|
|
def __init__(self, group_rows, result_rows, group_names):
|
|
self.group_rows = group_rows
|
|
self.result_rows = result_rows
|
|
self.group_names = group_names
|
|
self.kind = None
|
|
self.current_shop = None
|
|
self.windowed_results = False
|
|
self.group_limit = None
|
|
self.group_offset = 0
|
|
self.calls = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def execute(self, sql, params=()):
|
|
self.calls.append((sql, tuple(params)))
|
|
if 'COUNT(*) AS total' in sql:
|
|
self.kind = 'count'
|
|
elif 'MAX(t.created_at)' in sql:
|
|
self.kind = 'groups'
|
|
self.group_limit = int(params[-2])
|
|
self.group_offset = int(params[-1])
|
|
elif 'GROUP_CONCAT' in sql:
|
|
self.kind = 'group_names'
|
|
else:
|
|
self.kind = 'results'
|
|
self.windowed_results = 'ROW_NUMBER() OVER' in sql
|
|
self.current_shop = None if self.windowed_results else (str(params[-1]) if params else None)
|
|
|
|
def fetchone(self):
|
|
return {'total': len(self.group_rows)}
|
|
|
|
def fetchall(self):
|
|
if self.kind == 'groups':
|
|
end = self.group_offset + self.group_limit
|
|
return self.group_rows[self.group_offset:end]
|
|
if self.kind == 'group_names':
|
|
return self.group_names
|
|
if self.kind == 'results':
|
|
rows = [row for row in self.result_rows if str(row.get('result_file_url') or '').strip()]
|
|
if self.windowed_results:
|
|
counts = {}
|
|
limited = []
|
|
for row in rows:
|
|
shop_key = row['shop_name'].strip().casefold()
|
|
if counts.get(shop_key, 0) >= 3:
|
|
continue
|
|
counts[shop_key] = counts.get(shop_key, 0) + 1
|
|
limited.append(row)
|
|
return limited
|
|
if self.current_shop is not None:
|
|
rows = [row for row in rows if row['shop_name'].strip() == self.current_shop]
|
|
return rows[:3]
|
|
return []
|
|
|
|
|
|
class _FakeConnection:
|
|
def __init__(self, cursor):
|
|
self.cursor_value = cursor
|
|
|
|
def cursor(self):
|
|
return self.cursor_value
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class AdminShopDataGroupTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.app = Flask(__name__)
|
|
self.group_rows = [
|
|
{'shop_name': 'Shop A', 'latest_created_at': datetime(2026, 8, 5, 12, 0)},
|
|
{'shop_name': '', 'latest_created_at': datetime(2026, 8, 4, 12, 0)},
|
|
]
|
|
self.result_rows = [
|
|
self._result_row(6, 'Shop A', '2026-08-05T13:00:00', result_file_url=''),
|
|
self._result_row(5, 'Shop A', '2026-08-05T12:00:00'),
|
|
self._result_row(4, 'Shop A', '2026-08-04T12:00:00'),
|
|
self._result_row(3, 'Shop A', '2026-08-03T12:00:00'),
|
|
self._result_row(2, 'Shop A', '2026-08-02T12:00:00'),
|
|
self._result_row(1, '', '2026-08-04T11:00:00'),
|
|
]
|
|
|
|
@staticmethod
|
|
def _result_row(result_id, shop_name, created_at, result_file_url=None):
|
|
return {
|
|
'result_id': result_id,
|
|
'task_id': result_id + 100,
|
|
'user_id': 7,
|
|
'shop_name': shop_name,
|
|
'shop_id': shop_name.lower(),
|
|
'task_no': f'task-{result_id}',
|
|
'task_status': 'SUCCESS',
|
|
'result_success': 1,
|
|
'result_error': None,
|
|
'task_error': None,
|
|
'file_error': None,
|
|
'result_file_url': f'object-{result_id}' if result_file_url is None else result_file_url,
|
|
'result_filename': f'result-{result_id}.xlsx',
|
|
'result_file_size': 10,
|
|
'row_count': 2,
|
|
'request_json': '{}',
|
|
'created_at': created_at,
|
|
'updated_at': created_at,
|
|
'finished_at': created_at,
|
|
'file_job_id': None,
|
|
'file_status': 'SUCCESS',
|
|
'username': 'operator',
|
|
}
|
|
|
|
def test_group_item_caps_children_and_preserves_child_result_ids(self):
|
|
group = admin_api._shop_data_crawl_group_item(
|
|
{'shop_name': 'Shop A', 'latest_created_at': datetime(2026, 8, 5, 12, 0)},
|
|
{'shop a': [row for row in self.result_rows if row['result_file_url']][:4]},
|
|
{'shop a': 'Group 1'},
|
|
)
|
|
|
|
self.assertEqual(group['shop_name'], 'Shop A')
|
|
self.assertEqual(group['group_name'], 'Group 1')
|
|
self.assertEqual([item['result_id'] for item in group['results']], [5, 4, 3])
|
|
self.assertEqual(group['results'][0]['result_file_url'], 'object-5')
|
|
|
|
def test_list_paginates_groups_and_ignores_removed_user_status_filters(self):
|
|
cursor = _FakeCursor(
|
|
self.group_rows,
|
|
self.result_rows,
|
|
[{'shop_name': 'Shop A', 'group_name': 'Group 1'}],
|
|
)
|
|
connection = _FakeConnection(cursor)
|
|
with self.app.test_request_context(
|
|
'/api/admin/shop-data-crawl-tasks?page=1&page_size=10'
|
|
'&username=should-not-filter&status=FAILED&shop_name=Shop&group_name=Group'
|
|
'&created_from=2026-08-01T00:00'
|
|
):
|
|
with patch.object(admin_api, 'get_db', return_value=connection), \
|
|
patch.object(admin_api, '_ensure_backend_menu_access', return_value=(None, None, None)), \
|
|
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=(None, None, None)):
|
|
response = admin_api.list_shop_data_crawl_tasks.__wrapped__()
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.get_json()
|
|
self.assertEqual(body['items'], body['data']['items'])
|
|
payload = body['data']
|
|
self.assertEqual(payload['total'], 2)
|
|
self.assertEqual(payload['page'], 1)
|
|
self.assertEqual(payload['items'][0]['shop_name'], 'Shop A')
|
|
self.assertEqual(len(payload['items'][0]['results']), 3)
|
|
self.assertEqual(
|
|
[item['result_id'] for item in payload['items'][0]['results']],
|
|
[5, 4, 3],
|
|
)
|
|
self.assertEqual(payload['items'][1]['shop_name'], '未命名')
|
|
self.assertEqual(len(payload['items'][1]['results']), 1)
|
|
|
|
params = [param for _sql, call_params in cursor.calls for param in call_params]
|
|
self.assertNotIn('should-not-filter', params)
|
|
self.assertNotIn('FAILED', params)
|
|
self.assertTrue(any('GROUP BY TRIM(COALESCE(r.source_filename, ' in sql for sql, _ in cursor.calls))
|
|
self.assertTrue(any('ROW_NUMBER() OVER' in sql for sql, _ in cursor.calls))
|
|
self.assertTrue(any('shop_row_number <= 3' in sql for sql, _ in cursor.calls))
|
|
self.assertTrue(any('TRIM(COALESCE(sm.shop_name' in sql for sql, _ in cursor.calls))
|
|
self.assertTrue(any("TRIM(COALESCE(r.result_file_url, '')) <> ''" in sql for sql, _ in cursor.calls))
|
|
|
|
def test_list_returns_empty_items_when_group_page_is_out_of_range(self):
|
|
cursor = _FakeCursor(self.group_rows, self.result_rows, [])
|
|
connection = _FakeConnection(cursor)
|
|
with self.app.test_request_context(
|
|
'/api/admin/shop-data-crawl-tasks?page=2&page_size=10'
|
|
):
|
|
with patch.object(admin_api, 'get_db', return_value=connection), \
|
|
patch.object(admin_api, '_ensure_backend_menu_access', return_value=(None, None, None)), \
|
|
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=(None, None, None)):
|
|
response = admin_api.list_shop_data_crawl_tasks.__wrapped__()
|
|
|
|
payload = response.get_json()['data']
|
|
self.assertEqual(payload['total'], 2)
|
|
self.assertEqual(payload['page'], 2)
|
|
self.assertEqual(payload['items'], [])
|
|
self.assertFalse(any('ROW_NUMBER() OVER' in sql for sql, _ in cursor.calls))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|