169 lines
5.1 KiB
Python
169 lines
5.1 KiB
Python
"""
|
||
场景推荐接口测试
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from unittest.mock import patch, AsyncMock
|
||
from app.main import app
|
||
|
||
client = TestClient(app)
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_request_data():
|
||
"""示例请求数据"""
|
||
return {
|
||
"project_id": "project_001",
|
||
"company_info": {
|
||
"industry": ["零售", "电商"],
|
||
"description": "某连锁生鲜零售企业，主营水果、蔬菜等生鲜产品",
|
||
"data_scale": "100GB",
|
||
"data_sources": ["交易系统", "会员系统", "供应链系统"]
|
||
},
|
||
"data_assets": [
|
||
{
|
||
"name": "会员基础信息表",
|
||
"description": "存储C端注册用户的核心身份信息",
|
||
"core_tables": ["t_user_base_01", "t_user_profile_02"]
|
||
},
|
||
{
|
||
"name": "交易流水表",
|
||
"description": "记录所有交易订单的详细信息",
|
||
"core_tables": ["t_order_detail_01", "t_order_summary_02"]
|
||
}
|
||
],
|
||
"existing_scenarios": [
|
||
{
|
||
"name": "会员画像分析",
|
||
"description": "基于会员消费行为分析用户画像"
|
||
}
|
||
],
|
||
"options": {
|
||
"model": "qwen-max",
|
||
"temperature": 0.3
|
||
}
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_llm_response():
|
||
"""模拟大模型响应"""
|
||
return {
|
||
"recommended_scenarios": [
|
||
{
|
||
"scenario_name": "智能推荐系统",
|
||
"category": "营销增长",
|
||
"description": "基于用户历史行为和偏好，智能推荐商品",
|
||
"business_value": "提升转化率15%，增加客单价20%",
|
||
"data_requirements": ["会员基础信息表", "交易流水表"],
|
||
"priority": "高",
|
||
"estimated_effort": "中等",
|
||
"recommendation_score": 5
|
||
},
|
||
{
|
||
"scenario_name": "供应链优化",
|
||
"category": "降本增效",
|
||
"description": "优化库存管理，减少损耗",
|
||
"business_value": "降低库存成本10%，减少损耗5%",
|
||
"data_requirements": ["交易流水表", "供应链数据"],
|
||
"priority": "中",
|
||
"estimated_effort": "高",
|
||
"recommendation_score": 4
|
||
}
|
||
]
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scenario_recommendation_success(sample_request_data, mock_llm_response):
|
||
"""测试场景推荐成功"""
|
||
import json
|
||
with patch('app.services.scenario_recommendation_service.llm_client.call') as mock_call:
|
||
# 模拟大模型返回 JSON 字符串
|
||
mock_call.return_value = json.dumps(mock_llm_response, ensure_ascii=False)
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-recommendation",
|
||
json=sample_request_data
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["success"] is True
|
||
assert data["code"] == 200
|
||
assert "data" in data
|
||
assert "recommended_scenarios" in data["data"]
|
||
assert len(data["data"]["recommended_scenarios"]) > 0
|
||
|
||
|
||
def test_scenario_recommendation_request_validation():
|
||
"""测试请求验证"""
|
||
# 测试缺少必需字段
|
||
invalid_request = {
|
||
"project_id": "project_001"
|
||
}
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-recommendation",
|
||
json=invalid_request
|
||
)
|
||
|
||
assert response.status_code == 422 # 验证错误
|
||
|
||
|
||
def test_scenario_recommendation_empty_data_assets():
|
||
"""测试空数据资产列表"""
|
||
request_data = {
|
||
"project_id": "project_001",
|
||
"data_assets": [],
|
||
"existing_scenarios": []
|
||
}
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-recommendation",
|
||
json=request_data
|
||
)
|
||
|
||
# 应该返回 422 或 200（取决于业务逻辑）
|
||
assert response.status_code in [200, 422]
|
||
|
||
|
||
def test_scenario_recommendation_with_options():
|
||
"""测试带选项的请求"""
|
||
import json
|
||
request_data = {
|
||
"project_id": "project_001",
|
||
"company_info": {
|
||
"industry": ["零售"],
|
||
"description": "某连锁生鲜零售企业",
|
||
"data_scale": "100TB",
|
||
"data_sources": ["交易系统", "会员系统"]
|
||
},
|
||
"data_assets": [
|
||
{
|
||
"name": "测试表",
|
||
"description": "测试描述",
|
||
"core_tables": ["test_table_01"]
|
||
}
|
||
],
|
||
"existing_scenarios": [],
|
||
"options": {
|
||
"model": "gpt-4",
|
||
"temperature": 0.5
|
||
}
|
||
}
|
||
|
||
with patch('app.services.scenario_recommendation_service.llm_client.call') as mock_call:
|
||
mock_call.return_value = json.dumps({"recommended_scenarios": []}, ensure_ascii=False)
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-recommendation",
|
||
json=request_data
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|