"""
|
API接口测试
|
"""
|
import pytest
|
from fastapi.testclient import TestClient
|
from unittest.mock import patch
|
from main import app
|
|
|
client = TestClient(app)
|
|
|
class TestCallbackAPI:
|
"""回调API测试类"""
|
|
def test_health_check(self):
|
"""测试健康检查接口"""
|
response = client.get("/api/v1/health")
|
assert response.status_code == 200
|
data = response.json()
|
assert data["status"] == "healthy"
|
|
@patch('app.services.message_processor.message_processor')
|
@patch('app.workers.message_worker.message_worker')
|
def test_callback_success(self, mock_worker, mock_processor):
|
"""测试回调接口成功"""
|
mock_processor.enqueue_callback_message.return_value = True
|
|
callback_data = {
|
"account": "17200000000",
|
"messageType": "80001",
|
"wcId": "wxid_phyyedw9xap22",
|
"data": {
|
"content": "测试消息",
|
"fromGroup": "25411981800@chatroom",
|
"fromUser": "wxid_ynu1tgjz156j22",
|
"memberCount": 8,
|
"msgId": 1685074592,
|
"newMsgId": 1497474118261484795,
|
"self": False,
|
"timestamp": 1640772092,
|
"toUser": "wxid_phyyedw9xap22",
|
"wId": "12491ae9-62aa-4f7a-83e6-9db4e9f28e3c"
|
}
|
}
|
|
response = client.post("/api/v1/callback", json=callback_data)
|
|
assert response.status_code == 200
|
data = response.json()
|
assert data["success"] is True
|
assert data["message"] == "消息已成功加入处理队列"
|
|
mock_processor.enqueue_callback_message.assert_called_once()
|
mock_worker.process_user_queue.assert_called_once_with("wxid_ynu1tgjz156j22")
|
|
@patch('app.services.message_processor.message_processor')
|
def test_callback_failure(self, mock_processor):
|
"""测试回调接口失败"""
|
mock_processor.enqueue_callback_message.return_value = False
|
|
callback_data = {
|
"account": "17200000000",
|
"messageType": "80001",
|
"wcId": "wxid_phyyedw9xap22",
|
"data": {
|
"content": "测试消息",
|
"fromGroup": "25411981800@chatroom",
|
"fromUser": "wxid_ynu1tgjz156j22",
|
"self": False
|
}
|
}
|
|
response = client.post("/api/v1/callback", json=callback_data)
|
|
assert response.status_code == 200
|
data = response.json()
|
assert data["success"] is False
|
assert data["message"] == "消息处理失败"
|
|
def test_callback_invalid_data(self):
|
"""测试无效数据"""
|
invalid_data = {
|
"messageType": "80001"
|
# 缺少必要字段
|
}
|
|
response = client.post("/api/v1/callback", json=invalid_data)
|
assert response.status_code == 422 # 验证错误
|
|
def test_root_endpoint(self):
|
"""测试根路径"""
|
response = client.get("/")
|
assert response.status_code == 200
|
data = response.json()
|
assert data["message"] == "E云管家-DifyAI对接服务"
|
assert data["version"] == "1.0.0"
|
assert data["status"] == "running"
|
|
|
if __name__ == "__main__":
|
pytest.main([__file__])
|
