# 搜索引擎测试流水线指南 ## 概述 本文档介绍了搜索引擎项目的完整测试流水线,包括测试环境搭建、测试执行、结果分析等内容。测试流水线设计用于commit前的自动化质量保证。 ## 🏗️ 测试架构 ### 测试层次 ``` 测试流水线 ├── 代码质量检查 (Code Quality) │ ├── 代码格式化检查 (Black, isort) │ ├── 静态分析 (Flake8, MyPy, Pylint) │ └── 安全扫描 (Safety, Bandit) │ ├── 单元测试 (Unit Tests) │ ├── RequestContext测试 │ ├── Searcher测试 │ ├── QueryParser测试 │ └── BooleanParser测试 │ ├── 集成测试 (Integration Tests) │ ├── 端到端搜索流程测试 │ ├── 多组件协同测试 │ └── 错误处理测试 │ ├── API测试 (API Tests) │ ├── REST API接口测试 │ ├── 参数验证测试 │ ├── 并发请求测试 │ └── 错误响应测试 │ └── 性能测试 (Performance Tests) ├── 响应时间测试 ├── 并发性能测试 └── 资源使用测试 ``` ### 核心组件 1. **RequestContext**: 请求级别的上下文管理器,用于跟踪测试过程中的所有数据 2. **测试环境管理**: 自动化启动/停止测试依赖服务 3. **测试执行引擎**: 统一的测试运行和结果收集 4. **报告生成系统**: 多格式的测试报告生成 ## 🚀 快速开始 ### 本地测试环境 1. **启动测试环境** ```bash # 启动所有必要的测试服务 ./scripts/start_test_environment.sh ``` 2. **运行完整测试套件** ```bash # 运行所有测试 python scripts/run_tests.py # 或者使用pytest直接运行 pytest tests/ -v ``` 3. **停止测试环境** ```bash ./scripts/stop_test_environment.sh ``` ### CI/CD测试 1. **GitHub Actions** - Push到主分支自动触发 - Pull Request自动运行 - 手动触发支持 2. **测试报告** - 自动生成并上传 - PR评论显示测试摘要 - 详细报告下载 ## 📋 测试类型详解 ### 1. 单元测试 (Unit Tests) **位置**: `tests/unit/` **目的**: 测试单个函数、类、模块的功能 **覆盖范围**: - `test_context.py`: RequestContext功能测试 - `test_searcher.py`: Searcher核心功能测试 - `test_query_parser.py`: QueryParser处理逻辑测试 **运行方式**: ```bash # 运行所有单元测试 pytest tests/unit/ -v # 运行特定测试 pytest tests/unit/test_context.py -v # 生成覆盖率报告 pytest tests/unit/ --cov=. --cov-report=html ``` ### 2. 集成测试 (Integration Tests) **位置**: `tests/integration/` **目的**: 测试多个组件协同工作的功能 **覆盖范围**: - `test_search_integration.py`: 完整搜索流程集成 - 数据库、ES、搜索器集成测试 - 错误传播和处理测试 **运行方式**: ```bash # 运行集成测试(需要启动测试环境) pytest tests/integration/ -v -m "not slow" # 运行包含慢速测试的集成测试 pytest tests/integration/ -v ``` ### 3. API测试 (API Tests) **位置**: `tests/integration/test_api_integration.py` **目的**: 测试HTTP API接口的功能和性能 **覆盖范围**: - 基本搜索API - 参数验证 - 错误处理 - 并发请求 - Unicode支持 **运行方式**: ```bash # 运行API测试 pytest tests/integration/test_api_integration.py -v ``` ### 5. 索引 & 文档构建流水线验证(手动) 除了自动化测试外,推荐在联调/问题排查时手动跑一遍“**从 MySQL 到 ES doc**”的索引流水线,确保字段与 mapping、查询逻辑一致。 #### 5.1 启动 Indexer 服务 ```bash cd /home/tw/saas-search ./scripts/stop.sh # 停掉已有进程(可选) ./scripts/start_indexer.sh # 启动专用 indexer 服务,默认端口 6004 ``` #### 5.2 基于数据库构建 ES doc(只看、不写 ES) > 场景:已经知道某个 `tenant_id` 和 `spu_id`,想看它在“最新逻辑下”的 ES 文档长什么样。 ```bash curl -X POST "http://127.0.0.1:6004/indexer/build-docs-from-db" \ -H "Content-Type: application/json" \ -d '{ "tenant_id": "170", "spu_ids": ["223167"] }' ``` 返回中: - `docs[0]` 为当前代码构造出来的完整 ES doc(与 `mappings/search_products.json` 对齐); - 可以直接比对: - 索引字段说明:`docs/索引字段说明v2.md` - 实际 ES 文档:`docs/常用查询 - ES.md` 中的查询示例(按 `spu_id` 过滤)。 #### 5.3 与 ES 实际数据对比 ```bash curl -u 'essa:***' \ -X GET 'http://localhost:9200/search_products_tenant_170/_search?pretty' \ -H 'Content-Type: application/json' \ -d '{ "size": 5, "_source": ["title", "tags"], "query": { "bool": { "filter": [ { "term": { "spu_id": "223167" } } ] } } }' ``` 对比如下内容是否一致: - 多语言字段:`title/brief/description/vendor/category_name_text/category_path`; - 结构字段:`tags/specifications/skus/min_price/max_price/compare_at_price/total_inventory` 等; - 算法字段:`title_embedding` 是否存在(值不必逐项比对)。 如果两边不一致,可以结合: - `indexer/document_transformer.py`(文档构造逻辑); - `indexer/incremental_service.py`(增量索引/查库逻辑); - `logs/indexer.log`(索引日志) 逐步缩小问题范围。 ### 4. 性能测试 (Performance Tests) **目的**: 验证系统性能指标 **测试内容**: - 搜索响应时间 - API并发处理能力 - 资源使用情况 **运行方式**: ```bash # 运行性能测试 python scripts/run_performance_tests.py ``` ## 🛠️ 环境配置 ### 测试环境要求 1. **Python环境** ```bash # 创建测试环境 conda create -n searchengine-test python=3.9 conda activate searchengine-test # 安装依赖 pip install -r requirements.txt pip install pytest pytest-cov pytest-json-report ``` 2. **Elasticsearch** ```bash # 使用Docker启动ES docker run -d \ --name elasticsearch \ -p 9200:9200 \ -e "discovery.type=single-node" \ -e "xpack.security.enabled=false" \ elasticsearch:8.8.0 ``` 3. **环境变量** ```bash export ES_HOST="http://localhost:9200" export ES_USERNAME="elastic" export ES_PASSWORD="changeme" export API_HOST="127.0.0.1" export API_PORT="6003" export TENANT_ID="test_tenant" export TESTING_MODE="true" ``` ### 服务依赖 测试环境需要以下服务: 1. **Elasticsearch** (端口9200) - 存储和搜索测试数据 - 支持中文和英文索引 2. **API服务** (端口6003) - FastAPI测试服务 - 提供搜索接口 3. **测试数据库** - 预配置的测试索引 - 包含测试数据 ## 📊 测试报告 ### 报告类型 1. **实时控制台输出** - 测试进度显示 - 失败详情 - 性能摘要 2. **JSON格式报告** ```json { "timestamp": "2024-01-01T10:00:00", "summary": { "total_tests": 150, "passed": 148, "failed": 2, "success_rate": 98.7 }, "suites": { ... } } ``` 3. **文本格式报告** - 人类友好的格式 - 包含测试摘要和详情 - 适合PR评论 4. **HTML覆盖率报告** - 代码覆盖率可视化 - 分支和行覆盖率 - 缺失测试高亮 ### 报告位置 ``` test_logs/ ├── unit_test_results.json # 单元测试结果 ├── integration_test_results.json # 集成测试结果 ├── api_test_results.json # API测试结果 ├── test_report_20240101_100000.txt # 文本格式摘要 ├── test_report_20240101_100000.json # JSON格式详情 └── htmlcov/ # HTML覆盖率报告 ``` ## 🔄 CI/CD集成 ### GitHub Actions工作流 **触发条件**: - Push到主分支 - Pull Request创建/更新 - 手动触发 **工作流阶段**: 1. **代码质量检查** - 代码格式验证 - 静态代码分析 - 安全漏洞扫描 2. **单元测试** - 多Python版本矩阵测试 - 代码覆盖率收集 - 自动上传到Codecov 3. **集成测试** - 服务依赖启动 - 端到端功能测试 - 错误处理验证 4. **API测试** - 接口功能验证 - 参数校验测试 - 并发请求测试 5. **性能测试** - 响应时间检查 - 资源使用监控 - 性能回归检测 6. **测试报告生成** - 结果汇总 - 报告上传 - PR评论更新 ### 工作流配置 **文件**: `.github/workflows/test.yml` **关键特性**: - 并行执行提高效率 - 服务容器化隔离 - 自动清理资源 - 智能缓存依赖 ## 🧪 测试最佳实践 ### 1. 测试编写原则 - **独立性**: 每个测试应该独立运行 - **可重复性**: 测试结果应该一致 - **快速执行**: 单元测试应该快速完成 - **清晰命名**: 测试名称应该描述测试内容 ### 2. 测试数据管理 ```python # 使用fixture提供测试数据 @pytest.fixture def sample_tenant_config(): return TenantConfig( tenant_id="test_tenant", es_index_name="test_products" ) # 使用mock避免外部依赖 @patch('search.searcher.ESClient') def test_search_with_mock_es(mock_es_client, test_searcher): mock_es_client.search.return_value = mock_response result = test_searcher.search("test query") assert result is not None ``` ### 3. RequestContext集成 ```python def test_with_context(test_searcher): context = create_request_context("test-req", "test-user") result = test_searcher.search("test query", context=context) # 验证context被正确更新 assert context.query_analysis.original_query == "test query" assert context.get_stage_duration("elasticsearch_search") > 0 ``` ### 4. 性能测试指南 ```python def test_search_performance(client): start_time = time.time() response = client.get("/search", params={"q": "test query"}) response_time = (time.time() - start_time) * 1000 assert response.status_code == 200 assert response_time < 2000 # 2秒内响应 ``` ## 🚨 故障排除 ### 常见问题 1. **Elasticsearch连接失败** ```bash # 检查ES状态 curl http://localhost:9200/_cluster/health # 重启ES服务 docker restart elasticsearch ``` 2. **测试端口冲突** ```bash # 检查端口占用 lsof -i :6003 # 修改API端口 export API_PORT="6004" ``` 3. **依赖包缺失** ```bash # 重新安装依赖 pip install -r requirements.txt pip install pytest pytest-cov pytest-json-report ``` 4. **测试数据问题** ```bash # 重新创建测试索引 curl -X DELETE http://localhost:9200/test_products ./scripts/start_test_environment.sh ``` ### 调试技巧 1. **详细日志输出** ```bash pytest tests/unit/test_context.py -v -s --tb=long ``` 2. **运行单个测试** ```bash pytest tests/unit/test_context.py::TestRequestContext::test_create_context -v ``` 3. **调试模式** ```python import pdb; pdb.set_trace() ``` 4. **性能分析** ```bash pytest --profile tests/ ``` ## 📈 持续改进 ### 测试覆盖率目标 - **单元测试**: > 90% - **集成测试**: > 80% - **API测试**: > 95% ### 性能基准 - **搜索响应时间**: < 2秒 - **API并发处理**: 100 QPS - **系统资源使用**: < 80% CPU, < 4GB RAM ### 质量门禁 - **所有测试必须通过** - **代码覆盖率不能下降** - **性能不能显著退化** - **不能有安全漏洞**