""" 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__])