From 9cb45204c94fac46befcdfabff9673dc412cdd1c Mon Sep 17 00:00:00 2001
From: "mark.tian" <mark.tian@shiliulizi.cn>
Date: Tue, 2 Dec 2025 17:51:33 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=8E=A5=E5=8F=A3=E6=B5=8B?=
 =?UTF-8?q?=E8=AF=95=E5=8A=9F=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .env.example     |   2 +-
 TEST_README.md   | 156 ++++++++++++++++++++++++
 conftest.py      |  44 +++++++
 requirements.txt |   5 +-
 run_tests.py     |  77 ++++++++++++
 test_callback.py | 302 +++++++++++++++++++++++++++++++++++++++++++++++
 6 files changed, 584 insertions(+), 2 deletions(-)
 create mode 100644 TEST_README.md
 create mode 100644 conftest.py
 create mode 100644 run_tests.py
 create mode 100644 test_callback.py

diff --git a/.env.example b/.env.example
index 91cd2d4..65d473e 100644
--- a/.env.example
+++ b/.env.example
@@ -1,5 +1,5 @@
 # 数据库配置
-DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/ai_talk_callback_db
+DATABASE_URL=postgresql+asyncpg://postgres:12345@localhost:5432/ai_talk_callback_db
 
 # Redis配置
 REDIS_URL=redis://localhost:6379/0
diff --git a/TEST_README.md b/TEST_README.md
new file mode 100644
index 0000000..84e7048
--- /dev/null
+++ b/TEST_README.md
@@ -0,0 +1,156 @@
+# AI Talk Callback 接口测试说明
+
+## 概述
+
+本项目为 AI Talk 回调接口编写了完整的测试用例，覆盖了各种场景和边界条件。
+
+## 测试文件
+
+- `test_callback.py` - 主要测试文件，包含所有接口测试用例
+- `conftest.py` - pytest配置文件，包含测试fixtures
+- `run_tests.py` - 测试运行脚本
+
+## 测试覆盖的场景
+
+### 1. 基础功能测试
+- ✅ 根接口测试 (`GET /`)
+- ✅ 健康检查接口测试 (`GET /health`)
+
+### 2. 回调接口测试 (`POST /ai-talk/callback/{siteId}`)
+
+#### 2.1 正常业务流程
+- ✅ count < 3 且外部API调用成功
+- ✅ count >= 3 直接返回（不调用外部API）
+- ✅ 空数据列表处理
+
+#### 2.2 异常情况处理
+- ✅ 外部API调用失败
+- ✅ 数据库连接错误
+- ✅ Redis锁获取失败
+- ✅ 服务器内部错误
+
+#### 2.3 数据验证
+- ✅ 无效请求数据（缺少必需字段）
+- ✅ 负数count值验证
+
+## 运行测试
+
+### 方法1: 使用pytest直接运行
+```bash
+# 安装测试依赖
+pip install -r requirements.txt
+
+# 运行所有测试
+pytest test_callback.py -v
+
+# 运行特定测试
+pytest test_callback.py::TestAITalkCallback::test_callback_count_below_threshold_success -v
+
+# 运行并显示覆盖率
+pytest test_callback.py -v --cov=app --cov-report=html
+```
+
+### 方法2: 使用运行脚本
+```bash
+# 运行所有测试
+python run_tests.py
+
+# 运行特定测试
+python run_tests.py test_callback_count_below_threshold_success
+```
+
+## 测试数据
+
+测试使用了模拟的回调数据，包含完整的字段结构：
+
+```python
+{
+    "count": 2,
+    "data": [
+        {
+            "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": {
+                "number": "13800138000",
+                "province": "广东",
+                "city": "深圳",
+                "operator": "移动"
+            },
+            "group": {
+                "id": 1,
+                "name": "测试组"
+            },
+            "task": {
+                "id": "task-123",
+                "name": "测试任务"
+            },
+            "user": {
+                "id": "user-123",
+                "name": "测试用户"
+            },
+            "customer_data": {
+                "name": "测试客户",
+                "email": "test@example.com",
+                "company": "测试公司",
+                "extra": "额外信息"
+            }
+        }
+    ]
+}
+```
+
+## Mock策略
+
+测试中使用了以下Mock策略：
+
+1. **数据库Mock**: 使用 `AsyncMock` 模拟数据库会话
+2. **Redis Mock**: 模拟Redis分布式锁操作
+3. **外部API Mock**: 模拟外部API调用和重试机制
+4. **日志记录Mock**: 模拟数据库日志记录功能
+
+## 测试环境配置
+
+测试使用SQLite内存数据库，避免影响生产数据：
+
+```python
+TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
+```
+
+## 预期测试结果
+
+成功运行时应该看到类似输出：
+
+```
+============================= test session starts =============================
+collected 10 items
+
+test_callback.py::TestAITalkCallback::test_root_endpoint PASSED         [ 10%]
+test_callback.py::TestAITalkCallback::test_health_check PASSED          [ 20%]
+test_callback.py::TestAITalkCallback::test_callback_count_below_threshold_success PASSED [ 30%]
+test_callback.py::TestAITalkCallback::test_callback_count_above_threshold_direct_return PASSED [ 40%]
+test_callback.py::TestAITalkCallback::test_callback_external_api_failure PASSED [ 50%]
+test_callback.py::TestAITalkCallback::test_callback_invalid_request_data PASSED [ 60%]
+test_callback.py::TestAITalkCallback::test_callback_invalid_count_negative PASSED [ 70%]
+test_callback.py::TestAITalkCallback::test_callback_empty_data_list PASSED [ 80%]
+test_callback.py::TestAITalkCallback::test_callback_database_error PASSED [ 90%]
+test_callback.py::TestAITalkCallback::test_callback_redis_lock_error PASSED [100%]
+
+============================== 10 passed in 2.34s ==============================
+✅ 所有测试通过!
+```
+
+## 注意事项
+
+1. 确保安装了所有测试依赖
+2. 测试运行时会创建临时SQLite数据库文件
+3. 测试使用了Mock，不会实际调用外部API
+4. 建议在CI/CD环境中集成这些测试
\ No newline at end of file
diff --git a/conftest.py b/conftest.py
new file mode 100644
index 0000000..da3e6d5
--- /dev/null
+++ b/conftest.py
@@ -0,0 +1,44 @@
+import pytest
+import asyncio
+from unittest.mock import AsyncMock
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+
+# 测试数据库配置
+TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
+
+@pytest.fixture(scope="session")
+def event_loop():
+    """创建一个事件循环实例用于测试"""
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield loop
+    loop.close()
+
+@pytest.fixture
+async def test_engine():
+    """创建测试数据库引擎"""
+    from sqlalchemy.ext.asyncio import create_async_engine
+    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
+    yield engine
+    await engine.dispose()
+
+@pytest.fixture
+async def test_db_session(test_engine):
+    """创建测试数据库会话"""
+    from app.database import Base
+    
+    # 创建所有表
+    async with test_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    
+    # 创建会话
+    async_session = sessionmaker(
+        test_engine, class_=AsyncSession, expire_on_commit=False
+    )
+    
+    async with async_session() as session:
+        yield session
+    
+    # 清理：删除所有表
+    async with test_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index 43ebfbb..4afd73a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,4 +9,7 @@ pydantic==2.5.0
 pydantic-settings==2.1.0
 python-multipart==0.0.6
 httpx==0.25.2
-python-dotenv==1.0.0
\ No newline at end of file
+python-dotenv==1.0.0
+pytest==7.4.3
+pytest-asyncio==0.21.1
+aiosqlite==0.19.0
\ No newline at end of file
diff --git a/run_tests.py b/run_tests.py
new file mode 100644
index 0000000..ea9adba
--- /dev/null
+++ b/run_tests.py
@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+"""
+测试运行脚本
+"""
+
+import subprocess
+import sys
+import os
+
+def run_tests():
+    """运行所有测试"""
+    print("🧪 开始运行AI Talk回调接口测试...")
+    
+    # 设置环境变量为测试模式
+    os.environ["TESTING"] = "1"
+    
+    try:
+        # 运行pytest
+        result = subprocess.run([
+            sys.executable, "-m", "pytest", 
+            "test_callback.py", 
+            "-v",
+            "--tb=short",
+            "--color=yes"
+        ], capture_output=True, text=True)
+        
+        print(result.stdout)
+        if result.stderr:
+            print("错误信息:")
+            print(result.stderr)
+        
+        if result.returncode == 0:
+            print("✅ 所有测试通过!")
+            return True
+        else:
+            print("❌ 测试失败!")
+            return False
+            
+    except Exception as e:
+        print(f"❌ 运行测试时出错: {e}")
+        return False
+
+def run_specific_test(test_name: str):
+    """运行特定测试"""
+    print(f"🧪 运行特定测试: {test_name}")
+    
+    os.environ["TESTING"] = "1"
+    
+    try:
+        result = subprocess.run([
+            sys.executable, "-m", "pytest", 
+            "test_callback.py", 
+            f"-k={test_name}",
+            "-v",
+            "--tb=short",
+            "--color=yes"
+        ], capture_output=True, text=True)
+        
+        print(result.stdout)
+        if result.stderr:
+            print("错误信息:")
+            print(result.stderr)
+        
+        return result.returncode == 0
+        
+    except Exception as e:
+        print(f"❌ 运行测试时出错: {e}")
+        return False
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        # 运行特定测试
+        test_name = sys.argv[1]
+        run_specific_test(test_name)
+    else:
+        # 运行所有测试
+        run_tests()
\ No newline at end of file
diff --git a/test_callback.py b/test_callback.py
new file mode 100644
index 0000000..9745da1
--- /dev/null
+++ b/test_callback.py
@@ -0,0 +1,302 @@
+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"])
\ No newline at end of file