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 'AS latest_created_at' in sql and 'GROUP BY' 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) >= 1: 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[:1] 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, 7, 5, 21, 45)}, {'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, latest_file_updated_at=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, 'latest_file_updated_at': latest_file_updated_at or 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]) self.assertEqual(group['results'][0]['result_file_url'], 'object-5') def test_group_item_falls_back_to_daily_file_update_time(self): row = self._result_row( 21427, 'Shop A', '2026-08-06T15:20:43', latest_file_updated_at=datetime(2026, 8, 7, 5, 21, 45), ) group = admin_api._shop_data_crawl_group_item( {'shop_name': 'Shop A', 'latest_created_at': None}, {'shop a': [row]}, {}, ) self.assertEqual(group['latest_created_at'], '2026-08-07 05:21:45') 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(payload['items'][0]['latest_created_at'], '2026-08-07 05:21:45') self.assertEqual(len(payload['items'][0]['results']), 1) self.assertEqual( [item['result_id'] for item in payload['items'][0]['results']], [5], ) 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 <= 1' in sql for sql, _ in cursor.calls)) self.assertTrue(any( 'LEFT JOIN biz_shop_data_crawl_daily_file df ON df.latest_result_id = r.id' in sql for sql, _ in cursor.calls )) self.assertTrue(any( 'MAX(COALESCE(df.last_success_at, df.updated_at, t.finished_at, t.updated_at, t.created_at))' 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()