#!/usr/bin/env python3 """ 测试 Multi-Select Faceting 和 Selected 标记功能 验证： 1. multi_select=False 时，选中某个 facet 值后，该 facet 只返回一个值（标准模式） 2. multi_select=True 时，选中某个 facet 值后，该 facet 仍返回多个值（disjunctive 模式） 3. selected 字段正确标记 """ import requests import json API_URL = "http://localhost:8000/api/search" TENANT_ID = "test_tenant" def test_standard_faceting(): """测试标准 faceting（multi_select=False）""" print("\n" + "="*80) print("测试 1: 标准 Faceting (multi_select=False)") print("="*80) # 选择 category1_name = "玩具" payload = { "query": "*", "size": 5, "filters": { "category1_name": "玩具" }, "facets": [ { "field": "category1_name", "size": 10, "type": "terms", "multi_select": False # 标准模式 } ] } headers = {"X-Tenant-ID": TENANT_ID} try: response = requests.post(API_URL, json=payload, headers=headers) response.raise_for_status() result = response.json() print(f"\n✓ 请求成功 (HTTP {response.status_code})") print(f" 总结果数: {result.get('total', 0)}") if result.get('facets'): for facet in result['facets']: if facet['field'] == 'category1_name': print(f"\n Facet: {facet['field']}") print(f" Values 数量: {len(facet['values'])}") for val in facet['values']: selected_mark = "✓" if val.get('selected') else " " print(f" [{selected_mark}] {val['value']}: {val['count']}") # 验证 if len(facet['values']) == 1: print("\n ✓ 验证通过: multi_select=False 时，只返回选中的值") else: print("\n ⚠ 警告: multi_select=False 时应只返回一个值") selected_values = [v['value'] for v in facet['values'] if v.get('selected')] if selected_values == ['玩具']: print(" ✓ 验证通过: selected 字段正确标记") else: print(f" ✗ 错误: selected 标记不正确，期望 ['玩具']，实际 {selected_values}") else: print("\n ⚠ 警告: 没有返回 facets") except requests.exceptions.RequestException as e: print(f"\n✗ 请求失败: {e}") except Exception as e: print(f"\n✗ 错误: {e}") def test_multi_select_faceting(): """测试 Multi-Select Faceting (multi_select=True)""" print("\n" + "="*80) print("测试 2: Multi-Select Faceting (multi_select=True)") print("="*80) # 选择 category1_name = "玩具" payload = { "query": "*", "size": 5, "filters": { "category1_name": "玩具" }, "facets": [ { "field": "category1_name", "size": 10, "type": "terms", "multi_select": True # Multi-select 模式 } ] } headers = {"X-Tenant-ID": TENANT_ID} try: response = requests.post(API_URL, json=payload, headers=headers) response.raise_for_status() result = response.json() print(f"\n✓ 请求成功 (HTTP {response.status_code})") print(f" 总结果数: {result.get('total', 0)}") if result.get('facets'): for facet in result['facets']: if facet['field'] == 'category1_name': print(f"\n Facet: {facet['field']}") print(f" Values 数量: {len(facet['values'])}") for val in facet['values'][:5]: # 只显示前5个 selected_mark = "✓" if val.get('selected') else " " print(f" [{selected_mark}] {val['value']}: {val['count']}") # 验证 if len(facet['values']) > 1: print(f"\n ✓ 验证通过: multi_select=True 时，返回多个值 ({len(facet['values'])} 个)") else: print("\n ✗ 错误: multi_select=True 时应返回多个值") selected_values = [v['value'] for v in facet['values'] if v.get('selected')] if selected_values == ['玩具']: print(" ✓ 验证通过: selected 字段正确标记") else: print(f" ⚠ 警告: selected 标记可能不正确，期望 ['玩具']，实际 {selected_values}") else: print("\n ⚠ 警告: 没有返回 facets") except requests.exceptions.RequestException as e: print(f"\n✗ 请求失败: {e}") except Exception as e: print(f"\n✗ 错误: {e}") def test_specifications_multi_select(): """测试 Specifications 的 Multi-Select Faceting""" print("\n" + "="*80) print("测试 3: Specifications Multi-Select Faceting") print("="*80) # 选择 specifications.颜色 = "白色" payload = { "query": "*", "size": 5, "filters": { "specifications": { "name": "颜色", "value": "白色" } }, "facets": [ { "field": "specifications.颜色", "size": 10, "type": "terms", "multi_select": True }, { "field": "specifications.尺寸", "size": 10, "type": "terms", "multi_select": False } ] } headers = {"X-Tenant-ID": TENANT_ID} try: response = requests.post(API_URL, json=payload, headers=headers) response.raise_for_status() result = response.json() print(f"\n✓ 请求成功 (HTTP {response.status_code})") print(f" 总结果数: {result.get('total', 0)}") if result.get('facets'): for facet in result['facets']: print(f"\n Facet: {facet['field']} (multi_select={facet.get('multi_select', 'N/A')})") print(f" Values 数量: {len(facet['values'])}") for val in facet['values'][:5]: selected_mark = "✓" if val.get('selected') else " " print(f" [{selected_mark}] {val['value']}: {val['count']}") # 验证 specifications.颜色 if facet['field'] == 'specifications.颜色': if len(facet['values']) > 1: print(f" ✓ 验证通过: multi_select=True，返回多个颜色选项") selected_values = [v['value'] for v in facet['values'] if v.get('selected')] if '白色' in selected_values: print(" ✓ 验证通过: '白色' 被正确标记为 selected") # 验证 specifications.尺寸（基于白色商品的尺寸分布） if facet['field'] == 'specifications.尺寸': print(f" ℹ 尺寸分布基于已选的颜色过滤器") else: print("\n ⚠ 警告: 没有返回 facets") except requests.exceptions.RequestException as e: print(f"\n✗ 请求失败: {e}") except Exception as e: print(f"\n✗ 错误: {e}") def test_es_query_structure(): """测试 ES Query 结构（需要 debug=True）""" print("\n" + "="*80) print("测试 4: ES Query 结构验证 (debug=True)") print("="*80) payload = { "query": "手机", "size": 1, "filters": { "category1_name": "电子产品", "specifications": {"name": "颜色", "value": "白色"} }, "facets": [ { "field": "category1_name", "size": 5, "multi_select": True }, { "field": "specifications.颜色", "size": 5, "multi_select": True } ], "debug": True } headers = {"X-Tenant-ID": TENANT_ID} try: response = requests.post(API_URL, json=payload, headers=headers) response.raise_for_status() result = response.json() print(f"\n✓ 请求成功") if result.get('debug_info') and result['debug_info'].get('es_query'): es_query = result['debug_info']['es_query'] # 检查 post_filter if 'post_filter' in es_query: print("\n ✓ ES Query 包含 post_filter:") print(f" {json.dumps(es_query['post_filter'], indent=4, ensure_ascii=False)[:200]}...") else: print("\n ℹ ES Query 不包含 post_filter（可能没有 multi-select 过滤器）") # 检查 query.bool.filter if 'query' in es_query and 'bool' in es_query['query']: filters = es_query['query']['bool'].get('filter', []) print(f"\n ✓ Query filters 数量: {len(filters)}") else: print("\n ⚠ 警告: debug_info 中没有 es_query") except requests.exceptions.RequestException as e: print(f"\n✗ 请求失败: {e}") except Exception as e: print(f"\n✗ 错误: {e}") if __name__ == "__main__": print("\n" + "="*80) print("Multi-Select Faceting 功能测试") print("="*80) print(f"\nAPI URL: {API_URL}") print(f"Tenant ID: {TENANT_ID}") # 运行测试 test_standard_faceting() test_multi_select_faceting() test_specifications_multi_select() test_es_query_structure() print("\n" + "="*80) print("测试完成") print("="*80)