tests/conftest.py

"""
pytest配置文件
提供测试夹具和共享配置
"""
import os
import sys
import pytest
import tempfile
from typing import Dict, Any, Generator
from unittest.mock import Mock, MagicMock
# 添加项目根目录到Python路径
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, project_root)
from config import SearchConfig, QueryConfig, IndexConfig, SPUConfig, FunctionScoreConfig, RerankConfig
from utils.es_client import ESClient
from search import Searcher
from query import QueryParser
from context import RequestContext, create_request_context
@pytest.fixture
def sample_index_config() -> IndexConfig:
    """样例索引配置"""
    return IndexConfig(
        name="default",
        label="默认索引",
        fields=["title.zh", "brief.zh", "tags"],
        boost=1.0
    )
@pytest.fixture
def sample_search_config(sample_index_config) -> SearchConfig:
    """样例搜索配置"""
    query_config = QueryConfig(
        enable_query_rewrite=True,
        enable_text_embedding=True,
        supported_languages=["zh", "en"]
    )
    spu_config = SPUConfig(
        enabled=True,
        spu_field="spu_id",
        inner_hits_size=3
    )
    function_score_config = FunctionScoreConfig()
    rerank_config = RerankConfig()
    return SearchConfig(
        es_index_name="test_products",
        field_boosts={
            "tenant_id": 1.0,
            "title.zh": 3.0,
            "brief.zh": 1.5,
            "tags": 1.0,
            "category_path.zh": 1.5,
        },
        indexes=[sample_index_config],
        query_config=query_config,
        function_score=function_score_config,
        rerank=rerank_config,
        spu_config=spu_config
    )
@pytest.fixture
def mock_es_client() -> Mock:
    """模拟ES客户端"""
    mock_client = Mock(spec=ESClient)
    # 模拟搜索响应
    mock_response = {
        "hits": {
            "total": {"value": 10},
            "max_score": 2.5,
            "hits": [
                {
                    "_id": "1",
                    "_score": 2.5,
                    "_source": {
                        "title": {"zh": "红色连衣裙"},
                        "vendor": {"zh": "测试品牌"},
                        "min_price": 299.0,
                        "category_id": "1"
                    }
                },
                {
                    "_id": "2",
                    "_score": 2.2,
                    "_source": {
                        "title": {"zh": "蓝色连衣裙"},
                        "vendor": {"zh": "测试品牌"},
                        "min_price": 399.0,
                        "category_id": "1"
                    }
                }
            ]
        },
        "took": 15
    }
    mock_client.search.return_value = mock_response
    return mock_client
@pytest.fixture
def test_searcher(sample_search_config, mock_es_client) -> Searcher:
    """测试用Searcher实例"""
    return Searcher(
        es_client=mock_es_client,
        config=sample_search_config
    )
@pytest.fixture
def test_query_parser(sample_search_config) -> QueryParser:
    """测试用QueryParser实例"""
    return QueryParser(sample_search_config)
@pytest.fixture
def test_request_context() -> RequestContext:
    """测试用RequestContext实例"""
    return create_request_context("test-req-001", "test-user")
@pytest.fixture
def sample_search_results() -> Dict[str, Any]:
    """样例搜索结果"""
    return {
        "query": "红色连衣裙",
        "expected_total": 2,
        "expected_products": [
            {"title": "红色连衣裙", "min_price": 299.0},
            {"title": "蓝色连衣裙", "min_price": 399.0}
        ]
    }
@pytest.fixture
def temp_config_file() -> Generator[str, None, None]:
    """临时配置文件"""
    import tempfile
    import yaml
    config_data = {
        "es_index_name": "test_products",
        "field_boosts": {
            "title.zh": 3.0,
            "brief.zh": 1.5,
            "tags": 1.0,
            "category_path.zh": 1.5
        },
        "indexes": [
            {
                "name": "default",
                "label": "默认索引",
                "fields": ["title.zh", "brief.zh", "tags"],
                "boost": 1.0
            }
        ],
        "query_config": {
            "supported_languages": ["zh", "en"],
            "default_language": "zh",
            "enable_text_embedding": True,
            "enable_query_rewrite": True
        },
        "spu_config": {
            "enabled": True,
            "spu_field": "spu_id",
            "inner_hits_size": 3
        },
        "ranking": {
            "expression": "bm25() + 0.2*text_embedding_relevance()",
            "description": "Test ranking"
        },
        "function_score": {
            "score_mode": "sum",
            "boost_mode": "multiply",
            "functions": []
        },
        "rerank": {
            "rerank_window": 386
        }
    }
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
        yaml.dump(config_data, f)
        temp_file = f.name
    yield temp_file
    # 清理
    os.unlink(temp_file)
@pytest.fixture
def mock_env_variables(monkeypatch):
    """设置环境变量"""
    monkeypatch.setenv("ES_HOST", "http://localhost:9200")
    monkeypatch.setenv("ES_USERNAME", "elastic")
    monkeypatch.setenv("ES_PASSWORD", "changeme")
# 标记配置
pytest_plugins = []
# 标记定义
def pytest_configure(config):
    """配置pytest标记"""
    config.addinivalue_line(
        "markers", "unit: 单元测试"
    )
    config.addinivalue_line(
        "markers", "integration: 集成测试"
    )
    config.addinivalue_line(
        "markers", "api: API测试"
    )
    config.addinivalue_line(
        "markers", "e2e: 端到端测试"
    )
    config.addinivalue_line(
        "markers", "performance: 性能测试"
    )
    config.addinivalue_line(
        "markers", "slow: 慢速测试"
    )
# 测试数据
@pytest.fixture
def test_queries():
    """测试查询集合"""
    return [
        "红色连衣裙",
        "wireless bluetooth headphones",
        "手机 手机壳",
        "laptop AND (gaming OR professional)",
        "运动鞋 -价格:0-500"
    ]
@pytest.fixture
def expected_response_structure():
    """期望的API响应结构"""
    return {
        "hits": list,
        "total": int,
        "max_score": float,
        "took_ms": int,
        "aggregations": dict,
        "query_info": dict,
        "performance_summary": dict
    }