150 lines
4.4 KiB
Python
150 lines
4.4 KiB
Python
"""
|
||
AI 分析接口测试
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from unittest.mock import patch
|
||
from app.main import app
|
||
from app.schemas.inventory import TableInput, FieldInput
|
||
|
||
client = TestClient(app)
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_request_data():
|
||
"""示例请求数据"""
|
||
return {
|
||
"tables": [
|
||
{
|
||
"raw_name": "t_user_base_01",
|
||
"fields": [
|
||
{
|
||
"raw_name": "user_id",
|
||
"type": "varchar(64)",
|
||
"comment": "用户ID"
|
||
},
|
||
{
|
||
"raw_name": "phone",
|
||
"type": "varchar(11)",
|
||
"comment": "手机号"
|
||
},
|
||
{
|
||
"raw_name": "id_card",
|
||
"type": "varchar(18)",
|
||
"comment": "身份证号"
|
||
}
|
||
]
|
||
}
|
||
],
|
||
"project_id": "project_001",
|
||
"industry": "retail-fresh",
|
||
"context": "某连锁生鲜零售企业，主营水果、蔬菜等生鲜产品",
|
||
"options": {
|
||
"model": "qwen-max",
|
||
"temperature": 0.3,
|
||
"enable_pii_detection": True,
|
||
"enable_important_data_detection": True
|
||
}
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_llm_response():
|
||
"""模拟大模型响应"""
|
||
return {
|
||
"tables": [
|
||
{
|
||
"raw_name": "t_user_base_01",
|
||
"ai_name": "会员基础信息表",
|
||
"desc": "存储C端注册用户的核心身份信息",
|
||
"confidence": 98,
|
||
"fields": [
|
||
{
|
||
"raw_name": "user_id",
|
||
"ai_name": "用户ID",
|
||
"desc": "用户的唯一标识符",
|
||
"pii": [],
|
||
"pii_type": None,
|
||
"is_important_data": False,
|
||
"confidence": 95
|
||
},
|
||
{
|
||
"raw_name": "phone",
|
||
"ai_name": "手机号",
|
||
"desc": "用户的联系电话",
|
||
"pii": ["手机号"],
|
||
"pii_type": "contact",
|
||
"is_important_data": False,
|
||
"confidence": 98
|
||
},
|
||
{
|
||
"raw_name": "id_card",
|
||
"ai_name": "身份证号",
|
||
"desc": "用户的身份证号码",
|
||
"pii": ["身份证号"],
|
||
"pii_type": "identity",
|
||
"is_important_data": False,
|
||
"confidence": 99
|
||
}
|
||
]
|
||
}
|
||
]
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ai_analyze_success(sample_request_data, mock_llm_response):
|
||
"""测试 AI 分析成功"""
|
||
import json
|
||
with patch('app.services.ai_analyze_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/inventory/ai-analyze",
|
||
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 "tables" in data["data"]
|
||
assert len(data["data"]["tables"]) > 0
|
||
|
||
|
||
def test_ai_analyze_request_validation():
|
||
"""测试请求验证"""
|
||
# 测试缺少必需字段
|
||
invalid_request = {
|
||
"tables": [],
|
||
"project_id": "project_001"
|
||
}
|
||
|
||
response = client.post(
|
||
"/api/v1/inventory/ai-analyze",
|
||
json=invalid_request
|
||
)
|
||
|
||
assert response.status_code == 422 # 验证错误
|
||
|
||
|
||
def test_ai_analyze_empty_tables():
|
||
"""测试空表列表"""
|
||
request_data = {
|
||
"tables": [],
|
||
"project_id": "project_001"
|
||
}
|
||
|
||
response = client.post(
|
||
"/api/v1/inventory/ai-analyze",
|
||
json=request_data
|
||
)
|
||
|
||
assert response.status_code == 422 # 验证错误（tables 最小长度要求）
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|