302 lines
9.5 KiB
Python
302 lines
9.5 KiB
Python
import pytest
|
|
import asyncio
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from main import app
|
|
from app.models import CallbackRequest, CallbackItem, NumberData, Group, Task, User, CustomerData
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
class TestAITalkCallback:
|
|
"""AI Talk回调接口测试用例"""
|
|
|
|
def setup_method(self):
|
|
"""每个测试方法执行前的设置"""
|
|
self.sample_callback_item = CallbackItem(
|
|
bill=100,
|
|
duration=60,
|
|
callid="test-call-123",
|
|
calldate="2024-01-01 12:00:00",
|
|
number="13800138000",
|
|
numberid="number-123",
|
|
customer_id="customer-123",
|
|
status=0,
|
|
status_str="失败",
|
|
user_id="user-123",
|
|
type=1,
|
|
number_data=NumberData(
|
|
number="13800138000",
|
|
province="广东",
|
|
city="深圳",
|
|
operator="移动"
|
|
),
|
|
group=Group(
|
|
id=1,
|
|
name="测试组"
|
|
),
|
|
task=Task(
|
|
id="task-123",
|
|
name="测试任务"
|
|
),
|
|
user=User(
|
|
id="user-123",
|
|
name="测试用户"
|
|
),
|
|
customer_data=CustomerData(
|
|
name="测试客户",
|
|
email="test@example.com",
|
|
company="测试公司",
|
|
extra="额外信息"
|
|
)
|
|
)
|
|
|
|
@pytest.fixture
|
|
def mock_db_session(self):
|
|
"""模拟数据库会话"""
|
|
return AsyncMock(spec=AsyncSession)
|
|
|
|
@pytest.fixture
|
|
def sample_callback_request(self):
|
|
"""示例回调请求数据"""
|
|
return CallbackRequest(
|
|
count=2,
|
|
data=[self.sample_callback_item]
|
|
)
|
|
|
|
@pytest.fixture
|
|
def sample_callback_request_high_count(self):
|
|
"""高count示例回调请求数据"""
|
|
return CallbackRequest(
|
|
count=5,
|
|
data=[self.sample_callback_item]
|
|
)
|
|
|
|
def test_root_endpoint(self):
|
|
"""测试根接口"""
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert "message" in response.json()
|
|
|
|
def test_health_check(self):
|
|
"""测试健康检查接口"""
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "healthy"}
|
|
|
|
@patch('app.routes.get_db')
|
|
@patch('app.routes.redis_manager')
|
|
@patch('app.routes.log_callback_request')
|
|
def test_callback_count_below_threshold_success(
|
|
self,
|
|
mock_log_callback,
|
|
mock_redis_manager,
|
|
mock_get_db,
|
|
sample_callback_request
|
|
):
|
|
"""测试count小于阈值且外部API调用成功的情况"""
|
|
# 设置模拟对象
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value = mock_db
|
|
|
|
# 模拟Redis锁
|
|
mock_lock = AsyncMock()
|
|
mock_lock.__aenter__ = AsyncMock(return_value=None)
|
|
mock_lock.__aexit__ = AsyncMock(return_value=None)
|
|
mock_redis_manager.create_lock.return_value = mock_lock
|
|
|
|
# 模拟外部API调用成功
|
|
with patch('app.routes.call_external_api_with_retry') as mock_api_call:
|
|
mock_api_call.return_value = (True, 0)
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-123",
|
|
json=sample_callback_request.model_dump()
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["processed"] is True
|
|
assert data["retry_count"] == 0
|
|
assert data["site_id"] == "test-site-123"
|
|
assert "外部API调用成功" in data["message"]
|
|
|
|
@patch('app.routes.get_db')
|
|
@patch('app.routes.redis_manager')
|
|
@patch('app.routes.log_callback_request')
|
|
def test_callback_count_above_threshold_direct_return(
|
|
self,
|
|
mock_log_callback,
|
|
mock_redis_manager,
|
|
mock_get_db,
|
|
sample_callback_request_high_count
|
|
):
|
|
"""测试count大于等于阈值直接返回的情况"""
|
|
# 设置模拟对象
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value = mock_db
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-456",
|
|
json=sample_callback_request_high_count.model_dump()
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["processed"] is False
|
|
assert data["retry_count"] == 0
|
|
assert data["site_id"] == "test-site-456"
|
|
assert "count=5 >= 3" in data["message"]
|
|
|
|
@patch('app.routes.get_db')
|
|
@patch('app.routes.redis_manager')
|
|
@patch('app.routes.log_callback_request')
|
|
def test_callback_external_api_failure(
|
|
self,
|
|
mock_log_callback,
|
|
mock_redis_manager,
|
|
mock_get_db,
|
|
sample_callback_request
|
|
):
|
|
"""测试外部API调用失败的情况"""
|
|
# 设置模拟对象
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value = mock_db
|
|
|
|
# 模拟Redis锁
|
|
mock_lock = AsyncMock()
|
|
mock_lock.__aenter__ = AsyncMock(return_value=None)
|
|
mock_lock.__aexit__ = AsyncMock(return_value=None)
|
|
mock_redis_manager.create_lock.return_value = mock_lock
|
|
|
|
# 模拟外部API调用失败
|
|
with patch('app.routes.call_external_api_with_retry') as mock_api_call:
|
|
mock_api_call.return_value = (False, 3)
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-789",
|
|
json=sample_callback_request.model_dump()
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
assert "外部API调用失败" in response.json()["detail"]
|
|
|
|
def test_callback_invalid_request_data(self):
|
|
"""测试无效的请求数据"""
|
|
# 测试缺少必需字段
|
|
invalid_request = {
|
|
"data": [] # 缺少count字段
|
|
}
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-invalid",
|
|
json=invalid_request
|
|
)
|
|
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
def test_callback_invalid_count_negative(self):
|
|
"""测试负数count值"""
|
|
invalid_request = {
|
|
"count": -1,
|
|
"data": []
|
|
}
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-negative",
|
|
json=invalid_request
|
|
)
|
|
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
@patch('app.routes.get_db')
|
|
@patch('app.routes.redis_manager')
|
|
@patch('app.routes.log_callback_request')
|
|
def test_callback_empty_data_list(
|
|
self,
|
|
mock_log_callback,
|
|
mock_redis_manager,
|
|
mock_get_db
|
|
):
|
|
"""测试空数据列表的情况"""
|
|
# 设置模拟对象
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value = mock_db
|
|
|
|
# 模拟Redis锁
|
|
mock_lock = AsyncMock()
|
|
mock_lock.__aenter__ = AsyncMock(return_value=None)
|
|
mock_lock.__aexit__ = AsyncMock(return_value=None)
|
|
mock_redis_manager.create_lock.return_value = mock_lock
|
|
|
|
# 模拟外部API调用成功
|
|
with patch('app.routes.call_external_api_with_retry') as mock_api_call:
|
|
mock_api_call.return_value = (True, 0)
|
|
|
|
request = CallbackRequest(count=0, data=[])
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-empty",
|
|
json=request.model_dump()
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["processed"] is True
|
|
|
|
@patch('app.routes.get_db')
|
|
@patch('app.routes.redis_manager')
|
|
@patch('app.routes.log_callback_request')
|
|
def test_callback_database_error(
|
|
self,
|
|
mock_log_callback,
|
|
mock_redis_manager,
|
|
mock_get_db,
|
|
sample_callback_request
|
|
):
|
|
"""测试数据库错误的情况"""
|
|
# 模拟数据库错误
|
|
mock_log_callback.side_effect = Exception("数据库连接失败")
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-db-error",
|
|
json=sample_callback_request.model_dump()
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
assert "服务器内部错误" in response.json()["detail"]
|
|
|
|
@patch('app.routes.get_db')
|
|
@patch('app.routes.redis_manager')
|
|
@patch('app.routes.log_callback_request')
|
|
def test_callback_redis_lock_error(
|
|
self,
|
|
mock_log_callback,
|
|
mock_redis_manager,
|
|
mock_get_db,
|
|
sample_callback_request
|
|
):
|
|
"""测试Redis锁错误的情况"""
|
|
# 设置模拟对象
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value = mock_db
|
|
|
|
# 模拟Redis锁错误
|
|
mock_redis_manager.create_lock.side_effect = Exception("Redis连接失败")
|
|
|
|
response = client.post(
|
|
"/ai-talk/callback/test-site-redis-error",
|
|
json=sample_callback_request.model_dump()
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
assert "服务器内部错误" in response.json()["detail"]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"]) |