231 lines
7.6 KiB
Python
231 lines
7.6 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 {
|
||
"existing_scenarios": [
|
||
{
|
||
"name": "会员画像分析",
|
||
"description": "基于会员消费行为分析用户画像"
|
||
}
|
||
],
|
||
"data_assets": [
|
||
{
|
||
"name": "会员基础信息表",
|
||
"description": "存储C端注册用户的核心身份信息",
|
||
"core_tables": ["t_user_base_01"]
|
||
}
|
||
],
|
||
"company_info": {
|
||
"industry": ["零售"],
|
||
"description": "某连锁生鲜零售企业"
|
||
}
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_request_data_with_screenshots():
|
||
"""示例请求数据（含截图）"""
|
||
return {
|
||
"existing_scenarios": [
|
||
{
|
||
"name": "会员画像分析",
|
||
"description": "基于会员消费行为分析用户画像"
|
||
}
|
||
],
|
||
"data_assets": [
|
||
{
|
||
"name": "会员基础信息表",
|
||
"description": "存储C端注册用户的核心身份信息",
|
||
"core_tables": ["t_user_base_01"]
|
||
}
|
||
],
|
||
"company_info": {
|
||
"industry": ["零售"],
|
||
"description": "某连锁生鲜零售企业"
|
||
},
|
||
"scenario_screenshots": [
|
||
"data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
|
||
]
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_llm_response():
|
||
"""模拟大模型响应"""
|
||
return {
|
||
"optimization_suggestions": [
|
||
{
|
||
"scenario_name": "会员画像分析",
|
||
"current_status": "当前仅基于交易数据进行画像，维度单一",
|
||
"suggestions": [
|
||
"增加用户行为轨迹分析，包括浏览、收藏、分享等",
|
||
"引入第三方数据源，丰富用户标签体系",
|
||
"优化画像可视化展示，提升用户体验"
|
||
],
|
||
"potential_value": "提升画像准确率20%，增加营销转化率15%"
|
||
}
|
||
]
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_vision_llm_response():
|
||
"""模拟视觉大模型响应"""
|
||
return """场景截图分析结果：
|
||
1. 界面布局：整体布局清晰，但信息密度较高，建议优化留白
|
||
2. 数据展示：图表类型单一，建议增加更多可视化方式
|
||
3. 交互体验：筛选功能不够直观，建议优化交互设计
|
||
4. 功能完整性：缺少导出功能，建议添加"""
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scenario_optimization_success(sample_request_data, mock_llm_response):
|
||
"""测试场景优化成功"""
|
||
import json
|
||
with patch('app.services.scenario_optimization_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-optimization",
|
||
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 "optimization_suggestions" in data["data"]
|
||
assert len(data["data"]["optimization_suggestions"]) > 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scenario_optimization_with_screenshots(sample_request_data_with_screenshots, mock_llm_response, mock_vision_llm_response):
|
||
"""测试场景优化（含截图）"""
|
||
import json
|
||
with patch('app.services.scenario_optimization_service.llm_client.call') as mock_call:
|
||
with patch('httpx.AsyncClient') as mock_httpx_client:
|
||
# 模拟大模型返回 JSON 字符串
|
||
mock_call.return_value = json.dumps(mock_llm_response, ensure_ascii=False)
|
||
|
||
# 模拟视觉大模型响应
|
||
mock_response = AsyncMock()
|
||
mock_response.status_code = 200
|
||
mock_response.raise_for_status = AsyncMock()
|
||
mock_response.json.return_value = {
|
||
"choices": [{
|
||
"message": {
|
||
"content": mock_vision_llm_response
|
||
}
|
||
}]
|
||
}
|
||
|
||
# 设置 AsyncClient 的上下文管理器
|
||
mock_client_instance = AsyncMock()
|
||
mock_client_instance.post = AsyncMock(return_value=mock_response)
|
||
mock_httpx_client.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance)
|
||
mock_httpx_client.return_value.__aexit__ = AsyncMock(return_value=False)
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-optimization",
|
||
json=sample_request_data_with_screenshots
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["success"] is True
|
||
assert "data" in data
|
||
|
||
|
||
def test_scenario_optimization_request_validation():
|
||
"""测试请求验证"""
|
||
# 测试空场景列表（应该是有效的，因为 existing_scenarios 是必需的但可以是空列表）
|
||
valid_request = {
|
||
"existing_scenarios": [],
|
||
"data_assets": [],
|
||
"company_info": {
|
||
"industry": ["零售"],
|
||
"description": "某连锁生鲜零售企业"
|
||
}
|
||
}
|
||
|
||
with patch('app.services.scenario_optimization_service.llm_client.call') as mock_call:
|
||
import json
|
||
mock_call.return_value = json.dumps({"optimization_suggestions": []}, ensure_ascii=False)
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-optimization",
|
||
json=valid_request
|
||
)
|
||
|
||
# 应该返回 200（空场景列表也可以处理）
|
||
assert response.status_code == 200
|
||
|
||
|
||
def test_scenario_optimization_empty_scenarios():
|
||
"""测试空场景列表"""
|
||
request_data = {
|
||
"existing_scenarios": [],
|
||
"data_assets": [],
|
||
"company_info": {
|
||
"industry": ["零售"],
|
||
"description": "某连锁生鲜零售企业"
|
||
}
|
||
}
|
||
|
||
with patch('app.services.scenario_optimization_service.llm_client.call') as mock_call:
|
||
import json
|
||
mock_call.return_value = json.dumps({"optimization_suggestions": []}, ensure_ascii=False)
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-optimization",
|
||
json=request_data
|
||
)
|
||
|
||
# 应该返回 200（空场景列表也可以处理）
|
||
assert response.status_code == 200
|
||
|
||
|
||
def test_scenario_optimization_with_options():
|
||
"""测试带选项的请求"""
|
||
import json
|
||
request_data = {
|
||
"existing_scenarios": [
|
||
{
|
||
"name": "测试场景",
|
||
"description": "测试描述"
|
||
}
|
||
],
|
||
"data_assets": [],
|
||
"options": {
|
||
"model": "gpt-4",
|
||
"temperature": 0.5
|
||
}
|
||
}
|
||
|
||
with patch('app.services.scenario_optimization_service.llm_client.call') as mock_call:
|
||
mock_call.return_value = json.dumps({"optimization_suggestions": []}, ensure_ascii=False)
|
||
|
||
response = client.post(
|
||
"/api/v1/value/scenario-optimization",
|
||
json=request_data
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|