012-kaopeilian/backend/tests/test_coze_service.py

"""
Coze 服务层单元测试
"""

import asyncio
import pytest
from datetime import datetime
from unittest.mock import Mock, AsyncMock, patch, MagicMock

from cozepy import ChatEventType

from app.services.ai.coze.service import CozeService, get_coze_service
from app.services.ai.coze.models import (
    SessionType, MessageRole, ContentType, StreamEventType,
    CreateSessionRequest, SendMessageRequest, EndSessionRequest,
    CozeSession, CozeMessage, StreamEvent
)
from app.services.ai.coze.exceptions import CozeAPIError


@pytest.fixture
def coze_service():
    """创建测试用的服务实例"""
    with patch("app.services.ai.coze.service.get_coze_client"):
        service = CozeService()
        service.bot_config = {
            "course_chat": "chat-bot-id",
            "training": "training-bot-id",
            "exam": "exam-bot-id"
        }
        service.workspace_id = "test-workspace"
        return service


@pytest.mark.asyncio
class TestCozeService:
    """测试 Coze 服务"""
    
    async def test_create_course_chat_session(self, coze_service):
        """测试创建课程对话会话"""
        # Mock Coze client
        mock_conversation = Mock(id="conv-123")
        coze_service.client.conversations.create = Mock(return_value=mock_conversation)
        
        request = CreateSessionRequest(
            session_type=SessionType.COURSE_CHAT,
            user_id="user-123",
            course_id="course-456"
        )
        
        response = await coze_service.create_session(request)
        
        # 验证结果
        assert response.conversation_id == "conv-123"
        assert response.bot_id == "chat-bot-id"
        assert isinstance(response.session_id, str)
        assert isinstance(response.created_at, datetime)
        
        # 验证会话已保存
        session = coze_service._sessions[response.session_id]
        assert session.session_type == SessionType.COURSE_CHAT
        assert session.user_id == "user-123"
        assert session.metadata["course_id"] == "course-456"
    
    async def test_create_training_session(self, coze_service):
        """测试创建陪练会话"""
        mock_conversation = Mock(id="conv-456")
        coze_service.client.conversations.create = Mock(return_value=mock_conversation)
        
        request = CreateSessionRequest(
            session_type=SessionType.TRAINING,
            user_id="user-789",
            training_topic="客诉处理"
        )
        
        response = await coze_service.create_session(request)
        
        assert response.conversation_id == "conv-456"
        assert response.bot_id == "training-bot-id"
        
        session = coze_service._sessions[response.session_id]
        assert session.session_type == SessionType.TRAINING
        assert session.metadata["training_topic"] == "客诉处理"
    
    async def test_send_message_with_stream(self, coze_service):
        """测试发送消息（流式响应）"""
        # 创建测试会话
        session = CozeSession(
            session_id="test-session",
            conversation_id="conv-123",
            session_type=SessionType.COURSE_CHAT,
            user_id="user-123",
            bot_id="chat-bot-id"
        )
        coze_service._sessions["test-session"] = session
        coze_service._messages["test-session"] = []
        
        # Mock 流式响应
        mock_events = [
            Mock(
                event=ChatEventType.CONVERSATION_MESSAGE_DELTA,
                conversation_id="conv-123",
                message=Mock(content="Hello ")
            ),
            Mock(
                event=ChatEventType.CONVERSATION_MESSAGE_DELTA,
                conversation_id="conv-123",
                message=Mock(content="world!")
            ),
            Mock(
                event=ChatEventType.CONVERSATION_MESSAGE_COMPLETED,
                conversation_id="conv-123",
                message=Mock(content="Hello world!"),
                usage={"tokens": 10}
            )
        ]
        
        coze_service.client.chat.stream = Mock(return_value=iter(mock_events))
        
        request = SendMessageRequest(
            session_id="test-session",
            content="Hi there",
            stream=True
        )
        
        # 收集事件
        events = []
        async for event in coze_service.send_message(request):
            events.append(event)
        
        # 验证事件
        assert len(events) == 4  # 2 delta + 1 completed + 1 done
        assert events[0].event == StreamEventType.MESSAGE_DELTA
        assert events[0].content == "Hello "
        assert events[1].event == StreamEventType.MESSAGE_DELTA
        assert events[1].content == "world!"
        assert events[2].event == StreamEventType.MESSAGE_COMPLETED
        assert events[2].content == "Hello world!"
        assert events[3].event == StreamEventType.DONE
        
        # 验证消息已保存
        messages = coze_service._messages["test-session"]
        assert len(messages) == 2  # 用户消息 + 助手消息
        assert messages[0].role == MessageRole.USER
        assert messages[0].content == "Hi there"
        assert messages[1].role == MessageRole.ASSISTANT
        assert messages[1].content == "Hello world!"
    
    async def test_send_message_error_handling(self, coze_service):
        """测试发送消息错误处理"""
        # 不存在的会话
        request = SendMessageRequest(
            session_id="nonexistent",
            content="Test"
        )
        
        with pytest.raises(CozeAPIError, match="会话不存在"):
            async for _ in coze_service.send_message(request):
                pass
    
    async def test_end_session(self, coze_service):
        """测试结束会话"""
        # 创建测试会话和消息
        created_at = datetime.now()
        session = CozeSession(
            session_id="test-session",
            conversation_id="conv-123",
            session_type=SessionType.TRAINING,
            user_id="user-123",
            bot_id="training-bot-id",
            created_at=created_at
        )
        coze_service._sessions["test-session"] = session
        coze_service._messages["test-session"] = [
            Mock(), Mock(), Mock()  # 3条消息
        ]
        
        request = EndSessionRequest(
            reason="用户主动结束",
            feedback={"rating": 5, "comment": "很有帮助"}
        )
        
        response = await coze_service.end_session("test-session", request)
        
        # 验证响应
        assert response.session_id == "test-session"
        assert isinstance(response.ended_at, datetime)
        assert response.message_count == 3
        assert response.duration_seconds > 0
        
        # 验证会话元数据
        assert session.metadata["end_reason"] == "用户主动结束"
        assert session.metadata["feedback"]["rating"] == 5
    
    async def test_end_nonexistent_session(self, coze_service):
        """测试结束不存在的会话"""
        request = EndSessionRequest()
        
        with pytest.raises(CozeAPIError, match="会话不存在"):
            await coze_service.end_session("nonexistent", request)
    
    async def test_get_session_messages(self, coze_service):
        """测试获取会话消息历史"""
        # 创建测试消息
        messages = [
            CozeMessage(
                message_id=f"msg-{i}",
                session_id="test-session",
                role=MessageRole.USER if i % 2 == 0 else MessageRole.ASSISTANT,
                content=f"Message {i}"
            )
            for i in range(10)
        ]
        coze_service._messages["test-session"] = messages
        
        # 测试分页
        result = await coze_service.get_session_messages("test-session", limit=5, offset=2)
        
        assert len(result) == 5
        assert result[0].content == "Message 2"
        assert result[4].content == "Message 6"
    
    async def test_stream_with_card_content(self, coze_service):
        """测试流式响应中的卡片内容"""
        # 创建测试会话
        session = CozeSession(
            session_id="test-session",
            conversation_id="conv-123",
            session_type=SessionType.EXAM,
            user_id="user-123",
            bot_id="exam-bot-id"
        )
        coze_service._sessions["test-session"] = session
        coze_service._messages["test-session"] = []
        
        # Mock 包含卡片的流式响应
        mock_events = [
            Mock(
                event=ChatEventType.CONVERSATION_MESSAGE_DELTA,
                conversation_id="conv-123",
                message=Mock(content='{"question": "测试题目"}', content_type="card")
            ),
            Mock(
                event=ChatEventType.CONVERSATION_MESSAGE_COMPLETED,
                conversation_id="conv-123",
                message=Mock(content='{"question": "测试题目"}', content_type="card")
            )
        ]
        
        coze_service.client.chat.stream = Mock(return_value=iter(mock_events))
        
        request = SendMessageRequest(
            session_id="test-session",
            content="生成一道考题"
        )
        
        events = []
        async for event in coze_service.send_message(request):
            events.append(event)
        
        # 验证卡片类型被正确识别
        assert events[0].content_type == ContentType.CARD
        assert events[1].content_type == ContentType.CARD
        
        # 验证消息保存时的内容类型
        messages = coze_service._messages["test-session"]
        assert messages[1].content_type == ContentType.CARD


def test_get_coze_service_singleton():
    """测试服务单例"""
    service1 = get_coze_service()
    service2 = get_coze_service()
    assert service1 is service2