后台店铺数据重复检查独立成菜单：删除店铺导入测试、新增撞款详情卡片区
- 删除「店铺导入测试」菜单及全部实现：admin.html 面板、admin.js 逻辑、 admin_api.py 路由/常量/权限映射、Java 默认菜单 - 「店铺数据记录」页内子 tab 移除，只保留店铺数据列表 - 「店铺管理」下新增「店铺数据重复检查」菜单（V106 导入测试菜单保留并新增 V107 清理 + 幂等插入新菜单，Java 清单同步） - 新面板：筛选/指标卡/店铺上架分布/矩阵表格/撞款详情卡片区（同 ASIN 跨店 上架明细，查看明细开抽屉，上架时间倒序） - 重复扫描结果改为 MySQL 缓存 + 每日定时扫描（start_duplicate_scan_scheduler） - 新增 test_admin_shop_data_duplicate_check.py 单元测试
This commit is contained in:
@@ -30,6 +30,7 @@ def _make_workbook(rows_by_sheet):
|
||||
class ShopDataDuplicateAsinTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.app = Flask(__name__)
|
||||
self.app.config['SECRET_KEY'] = 'test-secret'
|
||||
self.group_rows = [
|
||||
{'shop_name': 'Shop A', 'latest_created_at': datetime(2026, 8, 31, 1, 15, 1)},
|
||||
{'shop_name': 'Shop B', 'latest_created_at': datetime(2026, 8, 31, 1, 15, 2)},
|
||||
@@ -133,23 +134,38 @@ class ShopDataDuplicateAsinTest(unittest.TestCase):
|
||||
self.shop_files[shop_name].save(stream)
|
||||
return stream.getvalue()
|
||||
|
||||
def _run_request(self, query=''):
|
||||
cursor = self._FakeCursor(
|
||||
self.group_rows,
|
||||
[self._result_row(i + 1, row['shop_name'], self.shop_country[row['shop_name']])
|
||||
for i, row in enumerate(self.group_rows)],
|
||||
)
|
||||
connection = self._FakeConnection(cursor)
|
||||
def _build_cache(self):
|
||||
"""构造与旧扫描一致的缓存：3 家店铺、跨店重复 B0ABC111。"""
|
||||
return {
|
||||
'scanned_at': '2026-09-03 03:10:00',
|
||||
'summary': {'shop_count': 3, 'total': 1},
|
||||
'items': [{
|
||||
'asin': 'B0ABC111',
|
||||
'shop_count': 2,
|
||||
'record_count': 3,
|
||||
'occurrences': [
|
||||
{'asin': 'B0ABC111', 'date': '2026-08-30', 'price': 'GBP 9.99', 'brand': 'BrandA',
|
||||
'shop_name': 'Shop A', 'group_name': 'Group-Shop A', 'country_codes': ['UK', 'DE']},
|
||||
{'asin': 'B0ABC111', 'date': '2026-08-30', 'price': 'EUR 10.99', 'brand': 'BrandA',
|
||||
'shop_name': 'Shop A', 'group_name': 'Group-Shop A', 'country_codes': ['DE']},
|
||||
{'asin': 'B0ABC111', 'date': '2026-08-31', 'price': 'GBP 8.50', 'brand': 'BrandA',
|
||||
'shop_name': 'Shop B', 'group_name': 'Group-Shop B', 'country_codes': ['UK']},
|
||||
],
|
||||
}],
|
||||
}
|
||||
|
||||
def _run_request(self, query='', cache=None):
|
||||
# 界面默认读缓存；force 走实时扫描（需 mock 扫描核心）
|
||||
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-asins?' + query):
|
||||
with patch.object(admin_api, 'get_db', return_value=connection), \
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True), \
|
||||
patch.object(admin_api, '_latest_duplicate_scan', return_value=cache), \
|
||||
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)), \
|
||||
patch.object(admin_api, '_shop_data_crawl_fetch_result_bytes',
|
||||
side_effect=lambda row: self._make_workbook_bytes(row['shop_name'])):
|
||||
return admin_api.shop_data_crawl_duplicate_asins.__wrapped__()
|
||||
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=(None, None, None)):
|
||||
return admin_api.shop_data_crawl_duplicate_asins()
|
||||
|
||||
def test_detects_duplicate_asins_across_shops(self):
|
||||
response = self._run_request('page=1&page_size=10')
|
||||
response = self._run_request('page=1&page_size=10', cache=self._build_cache())
|
||||
self.assertEqual(response.status_code, 200)
|
||||
body = response.get_json()
|
||||
self.assertTrue(body['success'])
|
||||
@@ -165,44 +181,103 @@ class ShopDataDuplicateAsinTest(unittest.TestCase):
|
||||
shop_a = next(occ for occ in item['occurrences'] if occ['shop_name'] == 'Shop A')
|
||||
self.assertIn('UK', shop_a['country_codes'])
|
||||
self.assertEqual(shop_a['date'], '2026-08-30')
|
||||
self.assertEqual(body['scanned_at'], '2026-09-03 03:10:00')
|
||||
|
||||
def test_filters_asin_and_date_range(self):
|
||||
# asin 模糊：匹配 B0ABC222 的只有 Shop B 一家，不构成跨店重复 → total 0
|
||||
response = self._run_request('page=1&page_size=10&asin=B0ABC222')
|
||||
def test_no_cache_returns_pending_empty(self):
|
||||
# 无缓存时（定时任务尚未执行）：返回空 + pending 提示
|
||||
response = self._run_request('page=1&page_size=10', cache=None)
|
||||
body = response.get_json()
|
||||
self.assertTrue(body['success'])
|
||||
self.assertEqual(body['total'], 0)
|
||||
# 日期范围 08-31：B0ABC111 在 Shop B 有该日记录 → 命中，展示完整记录（跨店仍 2 家）
|
||||
response = self._run_request('page=1&page_size=10&date_from=2026-08-31&date_to=2026-08-31')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
self.assertEqual(body['items'][0]['asin'], 'B0ABC111')
|
||||
self.assertEqual(body['items'][0]['shop_count'], 2) # 完整记录仍跨店
|
||||
# 日期范围 08-30：Shop A 两条命中，同样返回完整记录
|
||||
response = self._run_request('page=1&page_size=10&date_from=2026-08-30T00:00&date_to=2026-08-30T23:59')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
|
||||
def test_filters_shop_and_country(self):
|
||||
# 店铺过滤 Shop B：命中含 Shop B 记录的 ASIN → B0ABC111，展示完整记录跨店
|
||||
response = self._run_request('page=1&page_size=10&shop_name=Shop+B')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
self.assertEqual(body['items'][0]['asin'], 'B0ABC111')
|
||||
# 国家过滤 FR：只有 Shop C 有 FR，其 ASIN 仅单店 → 不构成重复 → total 0
|
||||
response = self._run_request('page=1&page_size=10&country=FR')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 0)
|
||||
# 国家过滤 DE：Shop A 命中 → B0ABC111 完整记录 → total 1
|
||||
response = self._run_request('page=1&page_size=10&country=DE')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
self.assertEqual(body['items'][0]['asin'], 'B0ABC111')
|
||||
|
||||
def test_pagination_when_page_out_of_range(self):
|
||||
response = self._run_request('page=2&page_size=10')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
self.assertEqual(body['items'], [])
|
||||
self.assertTrue(body['pending'])
|
||||
|
||||
def test_cache_filters_match_live_semantics(self):
|
||||
cache = self._build_cache()
|
||||
# 店铺过滤 Shop B：命中 → 完整记录
|
||||
response = self._run_request('page=1&page_size=10&shop_name=Shop+B', cache=cache)
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
self.assertEqual(body['items'][0]['asin'], 'B0ABC111')
|
||||
# 日期范围 08-31 命中
|
||||
response = self._run_request('page=1&page_size=10&date_from=2026-08-31&date_to=2026-08-31', cache=cache)
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
# 国家 FR 无命中
|
||||
response = self._run_request('page=1&page_size=10&country=FR', cache=cache)
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 0)
|
||||
|
||||
def test_force_runs_live_scan(self):
|
||||
# force=1 触发实时扫描并保存缓存后返回
|
||||
cursor = self._FakeCursor(
|
||||
self.group_rows,
|
||||
[self._result_row(i + 1, row['shop_name'], self.shop_country[row['shop_name']])
|
||||
for i, row in enumerate(self.group_rows)],
|
||||
)
|
||||
connection = self._FakeConnection(cursor)
|
||||
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-asins?page=1&page_size=10&force=1'):
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True), \
|
||||
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)), \
|
||||
patch.object(admin_api, '_save_duplicate_scan', return_value=1), \
|
||||
patch.object(admin_api, '_latest_duplicate_scan',
|
||||
return_value=self._build_cache()), \
|
||||
patch.object(admin_api, '_shop_data_crawl_fetch_result_bytes',
|
||||
side_effect=lambda row: self._make_workbook_bytes(row['shop_name'])):
|
||||
response = admin_api.shop_data_crawl_duplicate_asins()
|
||||
self.assertIsNotNone(response)
|
||||
body = response.get_json()
|
||||
self.assertTrue(body['success'])
|
||||
self.assertEqual(body['total'], 1)
|
||||
self.assertEqual(body['items'][0]['asin'], 'B0ABC111')
|
||||
|
||||
def test_force_conflict_when_lock_held(self):
|
||||
# 锁被占（定时任务/他请求在扫）时 force 返回 409
|
||||
cursor = self._FakeCursor(self.group_rows[:0], [])
|
||||
connection = self._FakeConnection(cursor)
|
||||
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-asins?page=1&page_size=10&force=1'):
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True), \
|
||||
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)), \
|
||||
patch.object(admin_api, '_duplicate_scan_lock'):
|
||||
admin_api._duplicate_scan_lock.acquire.return_value = False
|
||||
response = admin_api.shop_data_crawl_duplicate_asins()
|
||||
# 409 返回 (jsonify, status) tuple
|
||||
self.assertEqual(response[1], 409)
|
||||
|
||||
def test_latest_duplicate_scan_deserializes_json_columns(self):
|
||||
"""真实读库：summary_json / payload_json 为 JSON 字符串，需反序列化为 dict/list。"""
|
||||
class _ScanCursor(self._FakeCursor):
|
||||
def __init__(self):
|
||||
self.kind = None
|
||||
|
||||
def execute(self, sql, params=()):
|
||||
self.kind = 'scan'
|
||||
|
||||
def fetchone(self):
|
||||
return {
|
||||
'id': 9,
|
||||
'summary_json': '{"shop_count": 3, "total": 1, "source": "import"}',
|
||||
'payload_json': '[{"asin": "B0ABC111", "shop_count": 2}]',
|
||||
'created_at': datetime(2026, 9, 3, 13, 35, 25),
|
||||
}
|
||||
|
||||
class _ScanConnection(self._FakeConnection):
|
||||
def cursor(self):
|
||||
return _ScanCursor()
|
||||
|
||||
with patch.object(admin_api, 'get_db', return_value=_ScanConnection(None)):
|
||||
cache = admin_api._latest_duplicate_scan()
|
||||
self.assertIsNotNone(cache)
|
||||
self.assertEqual(cache['summary']['shop_count'], 3)
|
||||
self.assertEqual(cache['summary']['source'], 'import')
|
||||
self.assertEqual(cache['items'][0]['asin'], 'B0ABC111')
|
||||
self.assertEqual(cache['scanned_at'], '2026-09-03 13:35:25')
|
||||
|
||||
def test_parse_workbook_skips_unknown_sheets(self):
|
||||
wb = _make_workbook({'英国': [('2026-08-30', 'B0TEST01', 'GBP 1.00', '')]})
|
||||
|
||||
@@ -0,0 +1,225 @@
|
||||
"""店铺数据重复检查接口单元测试：全量缓存格式、矩阵分页、主管/超管数据范围过滤。"""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from flask import Flask
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
from blueprints import admin_api
|
||||
|
||||
|
||||
def _shop(name, group, country_codes=None):
|
||||
return {
|
||||
'shop_name': name,
|
||||
'group_name': group,
|
||||
'country_codes': country_codes or [],
|
||||
'rows': [],
|
||||
}
|
||||
|
||||
|
||||
class DuplicateCheckApiTest(unittest.TestCase):
|
||||
"""直接对接口函数做单元测试：缓存通过 _latest_duplicate_scan mock 注入。"""
|
||||
|
||||
def setUp(self):
|
||||
self.app = Flask(__name__)
|
||||
self.app.config['SECRET_KEY'] = 'test-secret'
|
||||
# 4 家店 / 3 个组；ASIN 分布：
|
||||
# A1: ShopA(UK) + ShopB(UK) —— 跨店重复
|
||||
# B1: ShopA(DE) 唯一
|
||||
# C1: ShopC(FR) 唯一
|
||||
# D1: ShopD(UK) 唯一
|
||||
self.cache = {
|
||||
'scanned_at': '2026-09-04 03:10:00',
|
||||
'summary': {
|
||||
'shop_count': 4, 'asin_total': 4, 'record_total': 5,
|
||||
'duplicate_asin_total': 1, 'duplicate_shop_count': 2,
|
||||
'site_count': 3, 'asin_per_shop': 1.2, 'source': 'job',
|
||||
},
|
||||
'shops': [
|
||||
{'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['UK', 'DE'], 'asin_count': 2, 'record_count': 3},
|
||||
{'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'asin_count': 1, 'record_count': 1},
|
||||
{'shop_name': 'ShopC', 'group_name': 'GroupC', 'country_codes': ['FR'], 'asin_count': 1, 'record_count': 1},
|
||||
{'shop_name': 'ShopD', 'group_name': 'GroupD', 'country_codes': ['UK'], 'asin_count': 1, 'record_count': 1},
|
||||
],
|
||||
'items': [
|
||||
{'asin': 'A0000001', 'shop_count': 2, 'record_count': 2, 'occurrences': [
|
||||
{'asin': 'A0000001', 'date': '2026-08-30', 'price': 'GBP 9.99', 'brand': 'BrandA',
|
||||
'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['UK', 'DE'], 'country': 'UK'},
|
||||
{'asin': 'A0000001', 'date': '2026-08-31', 'price': 'GBP 8.50', 'brand': 'BrandA',
|
||||
'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'country': 'UK'},
|
||||
]},
|
||||
{'asin': 'B0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
|
||||
{'asin': 'B0000001', 'date': '2026-08-30', 'price': 'EUR 10.99', 'brand': 'BrandA',
|
||||
'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['DE'], 'country': 'DE'},
|
||||
]},
|
||||
{'asin': 'C0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
|
||||
{'asin': 'C0000001', 'date': '2026-08-29', 'price': 'EUR 7.50', 'brand': 'BrandC',
|
||||
'shop_name': 'ShopC', 'group_name': 'GroupC', 'country_codes': ['FR'], 'country': 'FR'},
|
||||
]},
|
||||
{'asin': 'D0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
|
||||
{'asin': 'D0000001', 'date': '2026-08-28', 'price': 'GBP 5.00', 'brand': 'BrandD',
|
||||
'shop_name': 'ShopD', 'group_name': 'GroupD', 'country_codes': ['UK'], 'country': 'UK'},
|
||||
]},
|
||||
],
|
||||
}
|
||||
|
||||
def _access_patches(self, role='super_admin', current_row=None):
|
||||
return [
|
||||
patch.object(admin_api, '_ensure_backend_menu_access',
|
||||
return_value=(role, current_row or {'id': 1}, None)),
|
||||
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
|
||||
return_value=(role, current_row or {'id': 1}, None)),
|
||||
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache),
|
||||
]
|
||||
|
||||
def _call(self, url, role='super_admin', current_row=None):
|
||||
with self.app.test_request_context(url):
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True):
|
||||
for p in self._access_patches(role, current_row):
|
||||
p.start()
|
||||
try:
|
||||
return admin_api.shop_data_crawl_duplicate_check_items()
|
||||
finally:
|
||||
for p in reversed(self._access_patches(role, current_row)):
|
||||
p.stop()
|
||||
|
||||
def test_overview_super_admin_sees_all(self):
|
||||
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-overview'):
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True), \
|
||||
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('super_admin', {'id': 1}, None)), \
|
||||
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('super_admin', {'id': 1}, None)), \
|
||||
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
|
||||
response = admin_api.shop_data_crawl_duplicate_check_overview()
|
||||
body = response.get_json()
|
||||
self.assertTrue(body['success'])
|
||||
self.assertFalse(body['pending'])
|
||||
self.assertEqual(body['summary']['asin_total'], 4)
|
||||
self.assertEqual(body['summary']['record_total'], 5)
|
||||
self.assertEqual(body['summary']['duplicate_asin_total'], 1)
|
||||
self.assertEqual(len(body['shops']), 4)
|
||||
|
||||
def test_items_matix_columns_are_shops(self):
|
||||
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=monitor')
|
||||
body = response.get_json()
|
||||
self.assertTrue(body['success'])
|
||||
self.assertEqual(body['total'], 1) # monitor 只保留跨店重复
|
||||
self.assertEqual(body['items'][0]['asin'], 'A0000001')
|
||||
self.assertEqual([shop['shop_name'] for shop in body['shops']],
|
||||
['ShopA', 'ShopB', 'ShopC', 'ShopD'])
|
||||
|
||||
def test_items_all_view_includes_unique_asins(self):
|
||||
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=all')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 4)
|
||||
|
||||
def test_items_filter_by_asin_and_site(self):
|
||||
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&asin=A0000001&site=UK')
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['total'], 1)
|
||||
self.assertEqual(body['items'][0]['asin'], 'A0000001')
|
||||
|
||||
def test_leader_sees_only_own_group_shops(self):
|
||||
# 主管 id=653 只管理 GroupA（含 ShopA/ShopB）
|
||||
with patch.object(admin_api, '_shop_data_managed_shop_names',
|
||||
return_value={'shopa', 'shopb'}):
|
||||
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=all',
|
||||
role='admin', current_row={'id': 653})
|
||||
body = response.get_json()
|
||||
self.assertEqual([shop['shop_name'] for shop in body['shops']], ['ShopA', 'ShopB'])
|
||||
self.assertEqual(body['total'], 2) # A0000001（跨店）+ B0000001（唯一）
|
||||
# 跨店 ASIN 在主管范围内仍是 2 家店
|
||||
item = next(i for i in body['items'] if i['asin'] == 'A0000001')
|
||||
self.assertEqual(item['shop_count'], 2)
|
||||
|
||||
def test_leader_overview_stats_recomputed_after_filter(self):
|
||||
# 主管见 2 家店：唯一ASIN 2、上架记录 3、重复ASIN 1、重复店铺 2
|
||||
with patch.object(admin_api, '_shop_data_managed_shop_names',
|
||||
return_value={'shopa', 'shopb'}):
|
||||
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-overview'):
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True), \
|
||||
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('admin', {'id': 653}, None)), \
|
||||
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('admin', {'id': 653}, None)), \
|
||||
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
|
||||
response = admin_api.shop_data_crawl_duplicate_check_overview()
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['summary']['shop_count'], 2)
|
||||
self.assertEqual(body['summary']['asin_total'], 2)
|
||||
self.assertEqual(body['summary']['record_total'], 3)
|
||||
self.assertEqual(body['summary']['duplicate_asin_total'], 1)
|
||||
|
||||
def test_leader_sees_nothing_when_no_group(self):
|
||||
with patch.object(admin_api, '_shop_data_managed_shop_names', return_value=set()):
|
||||
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=all',
|
||||
role='admin', current_row={'id': 999999})
|
||||
body = response.get_json()
|
||||
self.assertEqual(body['shops'], [])
|
||||
self.assertEqual(body['total'], 0)
|
||||
|
||||
def test_export_csv_contains_filtered_rows(self):
|
||||
"""导出 CSV：行=上架记录，含 BOM，按筛选裁剪。"""
|
||||
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-export?'):
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True), \
|
||||
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('super_admin', {'id': 1}, None)), \
|
||||
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('super_admin', {'id': 1}, None)), \
|
||||
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
|
||||
response = admin_api.shop_data_crawl_duplicate_check_export()
|
||||
self.assertEqual(response.status_code, 200)
|
||||
text = response.get_data(as_text=True)
|
||||
self.assertTrue(text.startswith(''))
|
||||
lines = text.lstrip('').strip().splitlines()
|
||||
self.assertEqual(lines[0], 'ASIN,店铺数,店铺,分组,站点,上架时间,价格,品牌')
|
||||
self.assertEqual(len(lines), 3) # header + 两条上架记录（A0000001 两店各一条）
|
||||
|
||||
def test_export_monitor_view_excludes_unique_asins(self):
|
||||
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-export?view=monitor'):
|
||||
with patch('utils.auth.session', {'user_id': 1}), \
|
||||
patch('utils.auth.is_session_user_valid', return_value=True), \
|
||||
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('super_admin', {'id': 1}, None)), \
|
||||
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('super_admin', {'id': 1}, None)), \
|
||||
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
|
||||
response = admin_api.shop_data_crawl_duplicate_check_export()
|
||||
lines = response.get_data(as_text=True).lstrip('').strip().splitlines()
|
||||
self.assertEqual(len(lines), 3)
|
||||
self.assertIn('A0000001', lines[1])
|
||||
|
||||
def test_internal_request_falls_back_to_system_operator(self):
|
||||
"""无请求上下文（定时扫描线程）时，内部请求用系统级超管作为 operatorId。"""
|
||||
with patch('utils.auth.session', {'user_id': 1}): # 仅用于模拟无异常环境
|
||||
# 无请求上下文：has_request_context() 为 False，直接走 _resolve_system_operator_id
|
||||
with patch.object(admin_api, '_resolve_internal_token', return_value='test-token'), \
|
||||
patch.object(admin_api, '_resolve_system_operator_id', return_value=7):
|
||||
headers, params = admin_api._backend_java_internal_request()
|
||||
self.assertEqual(headers.get('X-Internal-Token'), 'test-token')
|
||||
self.assertEqual(params, {'operatorId': 7})
|
||||
|
||||
def test_internal_request_system_operator_missing_fails(self):
|
||||
"""无请求上下文且系统中没有任何管理员时，直接报错而不是传 0。"""
|
||||
with patch.object(admin_api, '_resolve_internal_token', return_value='test-token'), \
|
||||
patch.object(admin_api, '_resolve_system_operator_id', return_value=None):
|
||||
with self.assertRaises(ValueError):
|
||||
admin_api._backend_java_internal_request()
|
||||
|
||||
def test_fetch_result_bytes_rejects_json_error_body(self):
|
||||
"""Java 内部端点返回 JSON 错误体（如鉴权失败）时，抛出业务错误而不是 BadZipFile。"""
|
||||
fake_response = Mock()
|
||||
fake_response.raise_for_status = Mock()
|
||||
fake_response.iter_content = Mock(return_value=[
|
||||
'{"success":false,"message":"用户不存在","data":null,"code":401}'.encode('utf-8')])
|
||||
fake_response.close = Mock()
|
||||
with patch.object(admin_api, '_backend_java_internal_request',
|
||||
return_value=({'X-Internal-Token': 'test-token'}, {'operatorId': 7})), \
|
||||
patch.object(admin_api, '_get_backend_java_session') as fake_session:
|
||||
fake_session.return_value.get = Mock(return_value=fake_response)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
admin_api._shop_data_crawl_fetch_result_bytes({'result_id': 123})
|
||||
self.assertIn('用户不存在', str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user