| """ |
| Claude Agent适配器扩展测试 |
| V2.2 重构 - 基于 claude-agent-sdk 的简化 API |
| """ |
| |
| import pytest |
| from unittest.mock import Mock, AsyncMock, patch, mock_open |
| import asyncio |
| import tempfile |
| import shutil |
| |
| import sys |
| from pathlib import Path |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) |
| |
| from claude_agent.telegram.claude_adapter import ClaudeAgentAdapter |
| |
| |
| class TestClaudeAgentAdapterExtended: |
| """Claude Agent适配器扩展测试 - V2.2""" |
| |
| @pytest.fixture |
| def mock_agent(self): |
| """创建Mock AgentCore""" |
| agent = Mock() |
| # Mock streaming response |
| async def mock_streaming_response(prompt): |
| yield "Mock" |
| yield " agent" |
| yield " response" |
| |
| agent.create_streaming_response = AsyncMock(side_effect=lambda *args: mock_streaming_response(args[0])) |
| agent.process_user_input = AsyncMock(return_value="Mock agent response") |
| agent.conversation_history = [] |
| agent.to_dict = Mock(return_value={'test': 'data'}) |
| agent.clean_conversation_history = Mock(return_value=0) |
| agent.trim_conversation_history = Mock(return_value=0) |
| agent.set_thinking_mode = Mock() |
| return agent |
| |
| @pytest.fixture |
| def adapter_with_mock(self, mock_agent): |
| """创建带Mock Agent的适配器""" |
| with patch('claude_agent.telegram.claude_adapter.PersistenceManager') as mock_persistence: |
| mock_persistence.return_value.load_agent_state.return_value = None |
| mock_persistence.return_value.save_agent_state.return_value = True |
| mock_persistence.return_value.save_conversation_history.return_value = True |
| mock_persistence.return_value.get_all_chat_ids.return_value = [] |
| |
| adapter = ClaudeAgentAdapter() |
| test_chat_id = 'test_chat' |
| adapter._agents[test_chat_id] = mock_agent |
| return adapter |
| |
| @pytest.fixture |
| def adapter_without_agent(self): |
| """创建没有预设Agent的适配器""" |
| with patch('claude_agent.telegram.claude_adapter.AgentCore') as mock_core: |
| mock_instance = Mock() |
| mock_instance.process_user_input = AsyncMock(return_value="Default agent response") |
| mock_instance.conversation_history = [] |
| mock_instance.to_dict = Mock(return_value={}) |
| mock_instance.clean_conversation_history = Mock(return_value=0) |
| mock_instance.trim_conversation_history = Mock(return_value=0) |
| mock_core.return_value = mock_instance |
| mock_core.from_dict.return_value = mock_instance |
| with patch('claude_agent.telegram.claude_adapter.PersistenceManager') as mock_pm: |
| mock_pm.return_value.get_all_chat_ids.return_value = [] |
| mock_pm.return_value.load_agent_state.return_value = None |
| return ClaudeAgentAdapter() |
| |
| @pytest.fixture |
| def sample_context(self): |
| """示例对话上下文""" |
| return [ |
| {'message': 'Hello', 'is_bot': False, 'timestamp': '2024-01-01T10:00:00'}, |
| {'message': 'Hi! How can I help?', 'is_bot': True, 'timestamp': '2024-01-01T10:00:01'}, |
| {'message': 'I need help', 'is_bot': False, 'timestamp': '2024-01-01T10:00:02'} |
| ] |
| |
| @pytest.fixture |
| def sample_user_info(self): |
| """示例用户信息""" |
| return { |
| 'username': 'testuser', |
| 'first_name': 'Test', |
| 'last_name': 'User', |
| 'language_code': 'en', |
| 'chat_type': 'private', |
| 'chat_id': 'test_chat' |
| } |
| |
| @pytest.mark.asyncio |
| async def test_process_message_success(self, mock_agent, sample_context, sample_user_info): |
| """测试成功处理消息""" |
| with patch('claude_agent.telegram.claude_adapter.PersistenceManager') as mock_persistence: |
| mock_persistence.return_value.load_agent_state.return_value = None |
| mock_persistence.return_value.save_agent_state.return_value = True |
| mock_persistence.return_value.save_conversation_history.return_value = True |
| mock_persistence.return_value.get_all_chat_ids.return_value = [] |
| |
| adapter = ClaudeAgentAdapter() |
| test_chat_id = 'test_chat' |
| adapter._agents[test_chat_id] = mock_agent |
| |
| message = "Test message" |
| result = await adapter.process_message(message, sample_context, sample_user_info) |
| |
| mock_agent.process_user_input.assert_called_once() |
| assert result == "Mock agent response" |
| |
| @pytest.mark.asyncio |
| async def test_process_message_agent_error(self, adapter_with_mock, mock_agent, sample_context, sample_user_info): |
| """测试Agent处理错误""" |
| mock_agent.process_user_input.side_effect = Exception("Agent error") |
| result = await adapter_with_mock.process_message("Test", sample_context, sample_user_info) |
| assert "主人" in result |
| |
| @pytest.mark.asyncio |
| async def test_process_with_image(self, adapter_with_mock, mock_agent, sample_context, sample_user_info): |
| """测试处理图片消息""" |
| message = "What's in this image?" |
| image_path = "/tmp/test_image.jpg" |
| |
| result = await adapter_with_mock.process_with_image(message, image_path, sample_context, sample_user_info) |
| |
| mock_agent.process_user_input.assert_called_once() |
| call_args = mock_agent.process_user_input.call_args[0][0] |
| assert image_path in call_args |
| assert message in call_args |
| assert "用户发送了一张图片" in call_args |
| assert result == "Mock agent response" |
| |
| @pytest.mark.asyncio |
| async def test_process_with_image_error(self, adapter_with_mock, mock_agent, sample_context, sample_user_info): |
| """测试处理图片消息时出错""" |
| mock_agent.process_user_input.side_effect = Exception("Image processing error") |
| result = await adapter_with_mock.process_with_image("Test", "/tmp/img.jpg", sample_context, sample_user_info) |
| assert "主人" in result |
| |
| @pytest.mark.asyncio |
| async def test_process_with_document(self, adapter_with_mock, mock_agent, sample_context, sample_user_info): |
| """测试处理文档消息""" |
| message = "Analyze this document" |
| document_path = "/tmp/test_doc.pdf" |
| |
| result = await adapter_with_mock.process_with_document(message, document_path, sample_context, sample_user_info) |
| |
| mock_agent.process_user_input.assert_called_once() |
| call_args = mock_agent.process_user_input.call_args[0][0] |
| assert document_path in call_args |
| assert message in call_args |
| assert "用户发送了一个文档" in call_args |
| assert result == "Mock agent response" |
| |
| @pytest.mark.asyncio |
| async def test_process_with_document_error(self, adapter_with_mock, mock_agent, sample_context, sample_user_info): |
| """测试处理文档消息时出错""" |
| mock_agent.process_user_input.side_effect = Exception("Document processing error") |
| result = await adapter_with_mock.process_with_document("Test", "/tmp/doc.pdf", sample_context, sample_user_info) |
| assert "主人" in result |
| |
| @pytest.mark.asyncio |
| async def test_process_yolo_mode(self, adapter_with_mock, mock_agent, sample_context, sample_user_info): |
| """测试YOLO模式处理""" |
| message = "Help me solve this complex problem" |
| result = await adapter_with_mock.process_yolo_mode(message, sample_context, sample_user_info) |
| |
| # 验证设置了YOLO模式 |
| mock_agent.set_thinking_mode.assert_called() |
| mock_agent.process_user_input.assert_called_once() |
| assert result == "Mock agent response" |
| |
| @pytest.mark.asyncio |
| async def test_process_yolo_mode_error(self, adapter_with_mock, mock_agent, sample_context, sample_user_info): |
| """测试YOLO模式处理错误""" |
| mock_agent.process_user_input.side_effect = Exception("YOLO processing error") |
| result = await adapter_with_mock.process_yolo_mode("Test", sample_context, sample_user_info) |
| assert "主人" in result |
| |
| def test_build_user_context_complete_info(self, adapter_with_mock, sample_user_info): |
| """测试构建用户上下文（完整信息）""" |
| result = adapter_with_mock._build_user_context(sample_user_info) |
| |
| assert "用户信息:" in result |
| assert "用户名: @testuser" in result |
| assert "名字: Test" in result |
| assert "姓氏: User" in result |
| assert "语言: en" in result |
| assert "这是私人对话" in result |
| |
| def test_build_user_context_minimal_info(self, adapter_with_mock): |
| """测试构建用户上下文（最小信息）""" |
| minimal_info = {'chat_type': 'group'} |
| result = adapter_with_mock._build_user_context(minimal_info) |
| assert "用户信息:" in result |
| assert "这是群组对话" in result |
| |
| def test_build_user_context_different_chat_types(self, adapter_with_mock): |
| """测试不同聊天类型的用户上下文""" |
| chat_types = { |
| 'group': '这是群组对话', |
| 'supergroup': '这是超级群组对话', |
| 'channel': '这是频道消息', |
| 'private': '这是私人对话' |
| } |
| |
| for chat_type, expected_text in chat_types.items(): |
| user_info = {'chat_type': chat_type} |
| result = adapter_with_mock._build_user_context(user_info) |
| assert expected_text in result |
| |
| def test_build_user_context_empty(self, adapter_with_mock): |
| """测试构建用户上下文（空信息）""" |
| result = adapter_with_mock._build_user_context({}) |
| assert result == "用户信息: 这是私人对话" |
| |
| def test_set_and_get_agent(self, adapter_with_mock, mock_agent): |
| """测试设置和获取Agent（per-chat模式）""" |
| chat_id = 'test_chat_123' |
| new_agent = Mock() |
| |
| # set_agent已废弃 |
| adapter_with_mock.set_agent(new_agent) |
| |
| # 测试获取特定chat的Agent |
| result = adapter_with_mock.get_agent(chat_id) |
| assert result is not None |
| |
| def test_initialization_with_default_agent(self, adapter_without_agent): |
| """测试使用默认Agent初始化""" |
| chat_id = 'test_chat' |
| agent = adapter_without_agent.get_agent(chat_id) |
| assert agent is not None |
| |
| @pytest.mark.asyncio |
| async def test_context_handling_edge_cases(self, adapter_with_mock, mock_agent): |
| """测试上下文处理边缘情况""" |
| user_info = {'chat_id': 'test_chat'} |
| |
| result = await adapter_with_mock.process_message("Test", [], user_info) |
| mock_agent.process_user_input.assert_called() |
| |
| def test_user_info_edge_cases(self, adapter_with_mock): |
| """测试用户信息边缘情况""" |
| user_info_with_none = { |
| 'username': None, |
| 'first_name': 'Test', |
| 'last_name': None, |
| 'language_code': '', |
| 'chat_type': 'private' |
| } |
| |
| result = adapter_with_mock._build_user_context(user_info_with_none) |
| assert "名字: Test" in result |
| |
| |
| class TestClaudeAgentAdapterCoverage: |
| """Claude Agent适配器覆盖率提升测试 - V2.2""" |
| |
| @pytest.fixture |
| def adapter_with_mocks(self): |
| """创建带Mock的适配器""" |
| with patch('claude_agent.telegram.claude_adapter.PersistenceManager') as mock_pm: |
| mock_persistence = Mock() |
| mock_persistence.load_agent_state = Mock(return_value=None) |
| mock_persistence.save_agent_state = Mock(return_value=True) |
| mock_persistence.save_conversation_history = Mock() |
| mock_persistence.load_conversation_history = Mock(return_value=[]) |
| mock_persistence.get_all_chat_ids = Mock(return_value=[]) |
| mock_persistence.cleanup_old_data = Mock(return_value=5) |
| mock_persistence.get_storage_stats = Mock(return_value={'total_conversations': 10}) |
| mock_pm.return_value = mock_persistence |
| |
| adapter = ClaudeAgentAdapter(storage_dir="temp/test", bot_id="test_bot") |
| adapter.persistence = mock_persistence |
| return adapter |
| |
| @pytest.fixture |
| def mock_agent(self): |
| """创建Mock Agent""" |
| agent = Mock() |
| agent.conversation_history = [] |
| agent.process_user_input = AsyncMock(return_value="Mock response") |
| agent.to_dict = Mock(return_value={'test': 'data'}) |
| agent.clean_conversation_history = Mock(return_value=0) |
| agent.trim_conversation_history = Mock(return_value=0) |
| agent.set_thinking_mode = Mock() |
| |
| async def mock_streaming(): |
| yield "Mock" |
| yield " response" |
| yield " stream" |
| |
| agent.create_streaming_response = AsyncMock(side_effect=lambda *args: mock_streaming()) |
| return agent |
| |
| def test_build_user_display_name_variations(self, adapter_with_mocks): |
| """测试构建用户显示名的各种情况""" |
| # 有用户名的情况 |
| user_info1 = {'username': 'testuser', 'first_name': 'Test'} |
| result1 = adapter_with_mocks._build_user_display_name(123, user_info1) |
| assert result1 == "@testuser" |
| |
| # 有完整姓名的情况 |
| user_info2 = {'first_name': 'Test', 'last_name': 'User'} |
| result2 = adapter_with_mocks._build_user_display_name(123, user_info2) |
| assert result2 == "Test User" |
| |
| # 只有first_name的情况 |
| user_info3 = {'first_name': 'Test'} |
| result3 = adapter_with_mocks._build_user_display_name(123, user_info3) |
| assert result3 == "Test" |
| |
| # 回退到ID的情况 |
| user_info4 = {} |
| result4 = adapter_with_mocks._build_user_display_name(123, user_info4) |
| assert result4 == "User123" |
| |
| def test_load_claude_md_scenarios(self, adapter_with_mocks): |
| """测试加载CLAUDE.md的各种场景""" |
| content = adapter_with_mocks._claude_md_content |
| assert "你是一个有用的AI助手" in content |
| |
| def test_get_telegram_fallback_response_types(self, adapter_with_mocks): |
| """测试不同类型的错误回退响应""" |
| user_input = "Test message" |
| |
| # 基础错误 |
| response1 = adapter_with_mocks._get_telegram_fallback_response(user_input, "Processing failed") |
| assert "主人" in response1 |
| |
| # API错误 |
| response2 = adapter_with_mocks._get_telegram_fallback_response(user_input, "403") |
| assert "主人" in response2 |
| |
| # 超时错误 |
| response3 = adapter_with_mocks._get_telegram_fallback_response(user_input, "timeout") |
| assert "主人" in response3 |
| |
| @pytest.mark.asyncio |
| async def test_save_message_to_agent_functionality(self, adapter_with_mocks, mock_agent): |
| """测试保存消息到Agent功能""" |
| chat_id = "test_chat" |
| user_id = 123 |
| message = "Test message" |
| is_bot = False |
| user_info = {'username': 'testuser', 'first_name': 'Test'} |
| |
| with patch.object(adapter_with_mocks, '_get_or_create_agent', return_value=mock_agent): |
| await adapter_with_mocks._save_message_to_agent(chat_id, user_id, message, is_bot, user_info) |
| adapter_with_mocks._get_or_create_agent.assert_called_once_with(chat_id) |
| |
| def test_restore_agents_functionality(self, adapter_with_mocks): |
| """测试恢复Agent功能""" |
| adapter_with_mocks._agents.clear() |
| |
| saved_states = { |
| "chat_123": {"conversation_history": [{"role": "user", "content": "Hello"}]}, |
| "chat_456": {"conversation_history": [{"role": "assistant", "content": "Hi"}]} |
| } |
| |
| adapter_with_mocks.persistence.load_agent_state = Mock( |
| side_effect=lambda chat_id: saved_states.get(chat_id) |
| ) |
| adapter_with_mocks.persistence.get_all_chat_ids = Mock( |
| return_value=list(saved_states.keys()) |
| ) |
| |
| mock_agent_instance = Mock() |
| mock_agent_instance.clean_conversation_history = Mock(return_value=0) |
| mock_agent_instance.trim_conversation_history = Mock(return_value=0) |
| |
| with patch('claude_agent.telegram.claude_adapter.AgentCore') as mock_agent_class: |
| mock_agent_class.from_dict.return_value = mock_agent_instance |
| |
| agent1 = adapter_with_mocks._get_or_create_agent("chat_123") |
| agent2 = adapter_with_mocks._get_or_create_agent("chat_456") |
| |
| assert len(adapter_with_mocks._agents) == 2 |
| |
| def test_get_storage_stats_functionality(self, adapter_with_mocks): |
| """测试获取存储统计功能""" |
| for i in range(3): |
| chat_id = f"stats_test_{i}" |
| adapter_with_mocks._agents[chat_id] = Mock() |
| |
| stats = adapter_with_mocks.get_storage_stats() |
| assert "active_agents" in stats |
| |
| @pytest.mark.asyncio |
| async def test_cleanup_old_conversations_functionality(self, adapter_with_mocks): |
| """测试清理旧对话功能""" |
| adapter_with_mocks.persistence.cleanup_old_data = Mock(return_value=5) |
| result = await adapter_with_mocks.cleanup_old_conversations(30) |
| assert result == 5 |
| |
| |
| class TestClaudeAgentAdapterPersistence: |
| """Claude Agent适配器持久化测试 - V2.2""" |
| |
| @pytest.fixture |
| def temp_storage_dir(self): |
| """创建临时存储目录""" |
| temp_dir = tempfile.mkdtemp() |
| yield temp_dir |
| shutil.rmtree(temp_dir) |
| |
| @pytest.fixture |
| def adapter_with_persistence(self, temp_storage_dir): |
| """创建带持久化的适配器""" |
| return ClaudeAgentAdapter(storage_dir=temp_storage_dir) |
| |
| @pytest.fixture |
| def sample_context(self): |
| """示例对话上下文""" |
| return [ |
| {'message': 'Hello', 'is_bot': False, 'timestamp': '2024-01-01T10:00:00'}, |
| {'message': 'Hi! How can I help?', 'is_bot': True, 'timestamp': '2024-01-01T10:00:01'}, |
| ] |
| |
| @pytest.fixture |
| def sample_user_info(self): |
| """示例用户信息""" |
| return { |
| 'username': 'testuser', |
| 'first_name': 'Test', |
| 'last_name': 'User', |
| 'language_code': 'en', |
| 'chat_type': 'private', |
| 'chat_id': 'test_chat_persistence' |
| } |
| |
| def test_adapter_initialization_with_persistence(self, temp_storage_dir): |
| """测试带持久化的适配器初始化""" |
| adapter = ClaudeAgentAdapter(storage_dir=temp_storage_dir) |
| |
| assert adapter.persistence is not None |
| assert adapter.persistence.storage_dir == Path(temp_storage_dir) |
| assert hasattr(adapter, '_agents') |
| |
| @pytest.mark.asyncio |
| async def test_streaming_response_persistence(self, adapter_with_persistence, sample_context, sample_user_info): |
| """测试流式响应的持久化""" |
| async def mock_streaming_generator(): |
| yield "First chunk" |
| yield "Second chunk" |
| yield "Final chunk" |
| |
| with patch.object(adapter_with_persistence, '_get_or_create_agent') as mock_get_agent: |
| mock_agent = Mock() |
| mock_agent.conversation_history = [] |
| mock_agent.to_dict.return_value = {'test': 'data'} |
| mock_agent.create_streaming_response.return_value = mock_streaming_generator() |
| mock_get_agent.return_value = mock_agent |
| |
| with patch.object(adapter_with_persistence.persistence, 'save_agent_state', return_value=True): |
| with patch.object(adapter_with_persistence.persistence, 'save_conversation_history', return_value=True): |
| |
| chunks = [] |
| async for chunk in adapter_with_persistence.create_streaming_response( |
| "Test message", sample_context, sample_user_info |
| ): |
| chunks.append(chunk) |
| |
| assert len(chunks) == 3 |
| |
| |
| class TestClaudeAdapterErrorHandling: |
| """测试错误处理场景 - V2.2""" |
| |
| @pytest.fixture |
| def basic_adapter(self): |
| """创建基础适配器""" |
| with patch('claude_agent.telegram.claude_adapter.PersistenceManager') as mock_pm: |
| mock_pm.return_value.get_all_chat_ids.return_value = [] |
| return ClaudeAgentAdapter() |
| |
| @pytest.fixture |
| def sample_user_info(self): |
| """样本用户信息""" |
| return { |
| 'user_id': 123456789, |
| 'username': 'testuser', |
| 'first_name': 'Test', |
| 'last_name': 'User', |
| 'chat_id': 'test_chat_error' |
| } |
| |
| @pytest.fixture |
| def sample_context(self): |
| """样本上下文""" |
| return [ |
| { |
| 'user_id': 123456789, |
| 'username': 'testuser', |
| 'message': 'Hello', |
| 'timestamp': '2023-01-01T00:00:00' |
| } |
| ] |
| |
| def test_get_agent_with_specific_chat_id(self, basic_adapter): |
| """测试获取特定聊天ID的Agent""" |
| chat_id = "test_chat_specific" |
| |
| with patch.object(basic_adapter, '_get_or_create_agent') as mock_create: |
| mock_agent = Mock() |
| mock_create.return_value = mock_agent |
| basic_adapter._agents[chat_id] = mock_agent |
| |
| result = basic_adapter.get_agent(chat_id) |
| assert result == mock_agent |
| |
| def test_get_agent_with_no_chat_id_and_no_default(self, basic_adapter): |
| """测试获取Agent时没有聊天ID且没有默认Agent""" |
| result = basic_adapter.get_agent() |
| assert result is None |
| |
| @pytest.mark.asyncio |
| async def test_cleanup_old_conversations_logic(self, basic_adapter): |
| """测试清理旧对话的逻辑""" |
| with patch.object(basic_adapter.persistence, 'cleanup_old_data', return_value=5): |
| result = await basic_adapter.cleanup_old_conversations(days_threshold=30) |
| assert result == 5 |
| |
| def test_get_storage_stats_details(self, basic_adapter): |
| """测试获取存储统计详情""" |
| mock_stats = { |
| 'agents_count': 10, |
| 'total_conversations': 100, |
| 'storage_size_mb': 25.5 |
| } |
| |
| with patch.object(basic_adapter.persistence, 'get_storage_stats', return_value=mock_stats): |
| result = basic_adapter.get_storage_stats() |
| assert 'active_agents' in result |
| |
| |
| class TestClaudeAdapterDeprecatedMethods: |
| """测试已弃用方法的覆盖率 - V2.2""" |
| |
| @pytest.fixture |
| def adapter_with_deprecated(self): |
| """创建包含已弃用方法的适配器""" |
| with patch('claude_agent.telegram.claude_adapter.PersistenceManager') as mock_pm: |
| mock_pm.return_value.get_all_chat_ids.return_value = [] |
| return ClaudeAgentAdapter() |
| |
| def test_set_agent_deprecated(self, adapter_with_deprecated): |
| """测试已弃用的set_agent方法""" |
| mock_agent = Mock() |
| adapter_with_deprecated.set_agent(mock_agent) |
| # 验证不会崩溃 |
| assert True |
| |
| |
| class TestClaudeAdapterUtilityMethods: |
| """测试实用工具方法 - V2.2""" |
| |
| @pytest.fixture |
| def utility_adapter(self): |
| """创建用于测试实用方法的适配器""" |
| with patch('claude_agent.telegram.claude_adapter.PersistenceManager') as mock_pm: |
| mock_pm.return_value.get_all_chat_ids.return_value = [] |
| return ClaudeAgentAdapter() |
| |
| def test_build_user_display_name_variations(self, utility_adapter): |
| """测试用户显示名的各种变化""" |
| # 有用户名的情况 |
| user_info_with_username = { |
| 'user_id': 123, |
| 'username': 'testuser', |
| 'first_name': 'Test', |
| 'last_name': 'User' |
| } |
| display_name = utility_adapter._build_user_display_name(123, user_info_with_username) |
| assert display_name == "@testuser" |
| |
| # 没有用户名但有姓名的情况 |
| user_info_with_names = { |
| 'user_id': 123, |
| 'first_name': 'Test', |
| 'last_name': 'User' |
| } |
| display_name = utility_adapter._build_user_display_name(123, user_info_with_names) |
| assert display_name == "Test User" |
| |
| # 只有first_name的情况 |
| user_info_first_only = { |
| 'user_id': 123, |
| 'first_name': 'Test' |
| } |
| display_name = utility_adapter._build_user_display_name(123, user_info_first_only) |
| assert display_name == "Test" |
| |
| # 什么都没有的情况 |
| user_info_minimal = {'user_id': 123} |
| display_name = utility_adapter._build_user_display_name(123, user_info_minimal) |
| assert display_name == "User123" |
| |
| def test_build_user_context_comprehensive(self, utility_adapter): |
| """测试构建用户上下文的综合情况""" |
| user_info = { |
| 'user_id': 123456789, |
| 'username': 'testuser', |
| 'first_name': 'Test', |
| 'last_name': 'User', |
| 'chat_id': 'test_chat' |
| } |
| |
| context = utility_adapter._build_user_context(user_info) |
| assert "用户名: @testuser" in context |
| assert "名字: Test" in context |
| assert "姓氏: User" in context |
| |
| def test_get_telegram_fallback_response_quota_error(self, utility_adapter): |
| """测试Telegram回退响应的配额错误""" |
| result = utility_adapter._get_telegram_fallback_response( |
| "Test input", "预扣费额度失败" |
| ) |
| assert "AI大脑暂时欠费了" in result |
| |
| def test_get_telegram_fallback_response_403_error(self, utility_adapter): |
| """测试Telegram回退响应的403错误""" |
| result = utility_adapter._get_telegram_fallback_response( |
| "Test input", "403" |
| ) |
| assert "AI大脑暂时欠费了" in result |
| |
| def test_get_telegram_fallback_response_timeout_error(self, utility_adapter): |
| """测试Telegram回退响应的超时错误""" |
| result = utility_adapter._get_telegram_fallback_response( |
| "Test input", "timeout" |
| ) |
| assert "大脑好像卡住了" in result or "卡住了一下下" in result |