test_search_with_source_fields.py
4.78 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
#!/usr/bin/env python3
"""
测试实际搜索功能中的source_fields应用
"""
import sys
import os
import json
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from config import ConfigLoader
def test_search_query_structure():
"""测试搜索查询是否正确应用了source_fields"""
print("测试搜索查询中的source_fields应用...")
try:
from search.searcher import Searcher
from utils.es_client import ESClient
# 加载配置
config_loader = ConfigLoader("config/schema")
config = config_loader.load_customer_config("customer1")
print(f"✓ 配置加载成功: {config.customer_id}")
print(f" source_fields配置数量: {len(config.query_config.source_fields)}")
# 创建ES客户端（使用模拟客户端避免实际连接）
class MockESClient:
def search(self, index_name, body, size=10, from_=0):
print(f"模拟ES搜索 - 索引: {index_name}")
print(f"查询body结构:")
print(json.dumps(body, indent=2, ensure_ascii=False))
# 检查_source配置
if "_source" in body:
print("✓ 查询包含_source配置")
source_config = body["_source"]
if "includes" in source_config:
print(f"✓ source includes字段: {source_config['includes']}")
return {
'took': 5,
'hits': {
'total': {'value': 0},
'max_score': 0.0,
'hits': []
}
}
else:
print("✗ _source配置中缺少includes")
return None
else:
print("✗ 查询中缺少_source配置")
return None
def client(self):
return self
# 创建Searcher实例
es_client = MockESClient()
searcher = Searcher(config, es_client)
print("\n测试文本搜索...")
result = searcher.search("test query", size=5)
if result:
print("✓ 文本搜索测试成功")
else:
print("✗ 文本搜索测试失败")
print("\n测试图像搜索...")
try:
result = searcher.search_by_image("http://example.com/image.jpg", size=3)
if result:
print("✓ 图像搜索测试成功")
else:
print("✗ 图像搜索测试失败")
except Exception as e:
print(f"✗ 图像搜索测试失败: {e}")
return True
except Exception as e:
print(f"✗ 搜索测试失败: {e}")
import traceback
traceback.print_exc()
return False
def test_es_query_builder_integration():
"""测试ES查询构建器的集成"""
print("\n测试ES查询构建器集成...")
try:
from search.es_query_builder import ESQueryBuilder
# 创建构建器，传入空的source_fields列表
builder = ESQueryBuilder(
index_name="test_index",
match_fields=["title", "content"],
source_fields=None # 测试空配置的情况
)
query = builder.build_query("test query")
if "_source" not in query:
print("✓ 空source_fields配置下，查询不包含_source过滤")
else:
print("⚠ 空source_fields配置下，查询仍然包含_source过滤")
# 测试非空配置
builder2 = ESQueryBuilder(
index_name="test_index",
match_fields=["title", "content"],
source_fields=["id", "title"]
)
query2 = builder2.build_query("test query")
if "_source" in query2 and "includes" in query2["_source"]:
print("✓ 非空source_fields配置下，查询正确包含_source过滤")
else:
print("✗ 非空source_fields配置下，查询缺少_source过滤")
return True
except Exception as e:
print(f"✗ 查询构建器集成测试失败: {e}")
return False
if __name__ == "__main__":
print("=" * 60)
print("搜索功能source_fields应用测试")
print("=" * 60)
success = True
# 运行所有测试
success &= test_es_query_builder_integration()
success &= test_search_query_structure()
print("\n" + "=" * 60)
if success:
print("✓ 所有测试通过！source_fields在搜索功能中正确应用。")
print("✓ ES现在只返回配置中指定的字段，减少了网络传输和响应大小。")
else:
print("✗ 部分测试失败，请检查实现。")
print("=" * 60)