346 lines
11 KiB
Python
346 lines
11 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"]
|
||
|
||
@patch('app.routes.get_db')
|
||
@patch('app.routes.redis_manager')
|
||
def test_callback_ip_fields_saved(
|
||
self,
|
||
mock_redis_manager,
|
||
mock_get_db,
|
||
sample_callback_request
|
||
):
|
||
"""测试remote_address和server_ip字段是否正确保存"""
|
||
from app.database import CallbackLog
|
||
|
||
# 设置模拟对象
|
||
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-ip-fields",
|
||
json=sample_callback_request.model_dump()
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
|
||
# 验证log_callback_request被调用
|
||
assert mock_db.add.called
|
||
assert mock_db.commit.called
|
||
|
||
# 获取传递给add的CallbackLog对象
|
||
call_args = mock_db.add.call_args[0][0]
|
||
assert isinstance(call_args, CallbackLog)
|
||
|
||
# 验证新字段存在（可能为None，因为测试环境）
|
||
assert hasattr(call_args, 'remote_address')
|
||
assert hasattr(call_args, 'server_ip')
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"]) |