沃巴蒂
ff22ed3a8b
本次同步 09-12 至 09-28 的代码变更，主要内容： 新增功能 - 修改密码：后端新增 ChangePasswordRequest DTO 与UserController/UserService 对应接口，前端新增 ChangePasswordDialog 并在 Header 接入入口 - 文件预览组件重构：file-preview 目录拆分出 VideoPreview（支持 blobUrl 直传）、 ImagePreview、AudioPreview，index.tsx 统一调度各类型预览器 - 新增 my_ops 算子模块：markitdown_parser、mineru_pdf_parser、text_quality_evaluator - 新增 tools/minio-file-server MinIO 文件服务 功能增强 - 数据清洗：cleaning_task_service/scheduler/routes 重构， data_juicer_executor 与 wrapper 增强，OperatorTable/LogsTable/TaskDetail 同步调整 - 数据标注：annotation sync.py、project.py 接口与schema 更新 - 数据采集、算子市场、首页与菜单路由相关页面调整 - DataJuicer 算子接入（useOperatorOperations、dj_ops_schema） 结构调整 - deepflow/ 更名为 deep-flow/（419 个文件重命名） - 新增 helm 部署编排、GitHub Actions 之外的自建镜像构建脚本 未纳入本次提交 - scripts-backup/、runtime/python-executor-backup/、components_backup/（备份目录） - *.backup*、__pycache__ 产物 - zh_core_web_sm-3.8.0-py3-none-any.whl（46MB 构建产物） - DataX/（内嵌独立仓库，远程亦未收录其源码）
264 lines
9.8 KiB
Python
264 lines
9.8 KiB
Python
import pytest
|
|
|
|
from src.tools.search_postprocessor import SearchResultPostProcessor
|
|
|
|
|
|
class TestSearchResultPostProcessor:
|
|
"""Test cases for SearchResultPostProcessor"""
|
|
|
|
@pytest.fixture
|
|
def post_processor(self):
|
|
"""Create a SearchResultPostProcessor instance for testing"""
|
|
return SearchResultPostProcessor(
|
|
min_score_threshold=0.5, max_content_length_per_page=100
|
|
)
|
|
|
|
def test_process_results_empty_input(self, post_processor):
|
|
"""Test processing empty results"""
|
|
results = []
|
|
processed = post_processor.process_results(results)
|
|
assert processed == []
|
|
|
|
def test_process_results_with_valid_page_results(self, post_processor):
|
|
"""Test processing valid page results"""
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Test Page",
|
|
"url": "https://example.com",
|
|
"content": "Test content",
|
|
"score": 0.8,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert processed[0]["title"] == "Test Page"
|
|
assert processed[0]["url"] == "https://example.com"
|
|
assert processed[0]["content"] == "Test content"
|
|
assert processed[0]["score"] == 0.8
|
|
|
|
def test_process_results_filter_low_score(self, post_processor):
|
|
"""Test filtering out low score results"""
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Low Score Page",
|
|
"url": "https://example.com/low",
|
|
"content": "Low score content",
|
|
"score": 0.3, # Below threshold of 0.5
|
|
},
|
|
{
|
|
"type": "page",
|
|
"title": "High Score Page",
|
|
"url": "https://example.com/high",
|
|
"content": "High score content",
|
|
"score": 0.9,
|
|
},
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert processed[0]["title"] == "High Score Page"
|
|
|
|
def test_process_results_remove_duplicates(self, post_processor):
|
|
"""Test removing duplicate URLs"""
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Page 1",
|
|
"url": "https://example.com",
|
|
"content": "Content 1",
|
|
"score": 0.8,
|
|
},
|
|
{
|
|
"type": "page",
|
|
"title": "Page 2",
|
|
"url": "https://example.com", # Duplicate URL
|
|
"content": "Content 2",
|
|
"score": 0.7,
|
|
},
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert processed[0]["title"] == "Page 1" # First one should be kept
|
|
|
|
def test_process_results_sort_by_score(self, post_processor):
|
|
"""Test sorting results by score in descending order"""
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Low Score",
|
|
"url": "https://example.com/low",
|
|
"content": "Low score content",
|
|
"score": 0.3,
|
|
},
|
|
{
|
|
"type": "page",
|
|
"title": "High Score",
|
|
"url": "https://example.com/high",
|
|
"content": "High score content",
|
|
"score": 0.9,
|
|
},
|
|
{
|
|
"type": "page",
|
|
"title": "Medium Score",
|
|
"url": "https://example.com/medium",
|
|
"content": "Medium score content",
|
|
"score": 0.6,
|
|
},
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 2 # Low score filtered out
|
|
# Should be sorted by score descending
|
|
assert processed[0]["title"] == "High Score"
|
|
assert processed[1]["title"] == "Medium Score"
|
|
|
|
def test_process_results_truncate_long_content(self, post_processor):
|
|
"""Test truncating long content"""
|
|
long_content = "A" * 150 # Longer than max_content_length of 100
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Long Content Page",
|
|
"url": "https://example.com",
|
|
"content": long_content,
|
|
"score": 0.8,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert len(processed[0]["content"]) == 103 # 100 + "..."
|
|
assert processed[0]["content"].endswith("...")
|
|
|
|
def test_process_results_remove_base64_images(self, post_processor):
|
|
"""Test removing base64 images from content"""
|
|
content_with_base64 = (
|
|
"Content with image "
|
|
+ "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg=="
|
|
)
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Page with Base64",
|
|
"url": "https://example.com",
|
|
"content": content_with_base64,
|
|
"score": 0.8,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert processed[0]["content"] == "Content with image "
|
|
|
|
def test_process_results_with_image_type(self, post_processor):
|
|
"""Test processing image type results"""
|
|
results = [
|
|
{
|
|
"type": "image",
|
|
"image_url": "https://example.com/image.jpg",
|
|
"image_description": "Test image",
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert processed[0]["type"] == "image"
|
|
assert processed[0]["image_url"] == "https://example.com/image.jpg"
|
|
assert processed[0]["image_description"] == "Test image"
|
|
|
|
def test_process_results_filter_base64_image_urls(self, post_processor):
|
|
"""Test filtering out image results with base64 URLs"""
|
|
results = [
|
|
{
|
|
"type": "image",
|
|
"image_url": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg==",
|
|
"image_description": "Base64 image",
|
|
},
|
|
{
|
|
"type": "image",
|
|
"image_url": "https://example.com/image.jpg",
|
|
"image_description": "Regular image",
|
|
},
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert processed[0]["image_url"] == "https://example.com/image.jpg"
|
|
|
|
def test_process_results_truncate_long_image_description(self, post_processor):
|
|
"""Test truncating long image descriptions"""
|
|
long_description = "A" * 150 # Longer than max_content_length of 100
|
|
results = [
|
|
{
|
|
"type": "image",
|
|
"image_url": "https://example.com/image.jpg",
|
|
"image_description": long_description,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert len(processed[0]["image_description"]) == 103 # 100 + "..."
|
|
assert processed[0]["image_description"].endswith("...")
|
|
|
|
def test_process_results_other_types_passthrough(self, post_processor):
|
|
"""Test that other result types pass through unchanged"""
|
|
results = [
|
|
{
|
|
"type": "video",
|
|
"title": "Test Video",
|
|
"url": "https://example.com/video.mp4",
|
|
"score": 0.8,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert processed[0]["type"] == "video"
|
|
assert processed[0]["title"] == "Test Video"
|
|
|
|
def test_process_results_truncate_long_content_with_no_config(self):
|
|
"""Test truncating long content"""
|
|
post_processor = SearchResultPostProcessor(None, None)
|
|
long_content = "A" * 150 # Longer than max_content_length of 100
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Long Content Page",
|
|
"url": "https://example.com",
|
|
"content": long_content,
|
|
"score": 0.8,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert len(processed[0]["content"]) == len("A" * 150)
|
|
|
|
def test_process_results_truncate_long_content_with_max_content_length_config(self):
|
|
"""Test truncating long content"""
|
|
post_processor = SearchResultPostProcessor(None, 100)
|
|
long_content = "A" * 150 # Longer than max_content_length of 100
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Long Content Page",
|
|
"url": "https://example.com",
|
|
"content": long_content,
|
|
"score": 0.8,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 1
|
|
assert len(processed[0]["content"]) == 103
|
|
assert processed[0]["content"].endswith("...")
|
|
|
|
def test_process_results_truncate_long_content_with_min_score_config(self):
|
|
"""Test truncating long content"""
|
|
post_processor = SearchResultPostProcessor(0.8, None)
|
|
long_content = "A" * 150 # Longer than max_content_length of 100
|
|
results = [
|
|
{
|
|
"type": "page",
|
|
"title": "Long Content Page",
|
|
"url": "https://example.com",
|
|
"content": long_content,
|
|
"score": 0.3,
|
|
}
|
|
]
|
|
processed = post_processor.process_results(results)
|
|
assert len(processed) == 0
|