| """ |
| SSHOUT模块简化测试 |
| 只测试基本导入和简单功能 |
| """ |
| |
| import pytest |
| from unittest.mock import Mock, patch, AsyncMock, MagicMock |
| import asyncio |
| from datetime import datetime |
| import struct |
| import paramiko |
| |
| import sys |
| from pathlib import Path |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) |
| |
| |
| class TestSSHOUTModuleBasics: |
| """SSHOUT模块基础测试""" |
| |
| def test_module_imports(self): |
| """测试模块能正常导入""" |
| try: |
| import claude_agent.sshout as sshout_module |
| assert sshout_module is not None |
| except ImportError: |
| pytest.fail("无法导入SSHOUT模块") |
| |
| def test_all_exports_exist(self): |
| """测试__all__导出的所有项目都存在""" |
| import claude_agent.sshout as sshout_module |
| |
| if hasattr(sshout_module, '__all__'): |
| for export_name in sshout_module.__all__: |
| assert hasattr(sshout_module, export_name), f"导出项 {export_name} 不存在" |
| |
| def test_create_integration_function_exists(self): |
| """测试create_sshout_integration函数存在""" |
| from claude_agent.sshout import create_sshout_integration |
| assert callable(create_sshout_integration) |
| |
| def test_create_integration_with_valid_config(self): |
| """测试使用有效配置创建集成""" |
| from claude_agent.sshout import create_sshout_integration |
| |
| mock_agent = Mock() |
| |
| # 正确的patch路径 |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'connection_mode': 'api', |
| 'api_endpoint': 'wss://example.com', |
| 'username': 'testuser', |
| 'password': 'testpass', |
| 'mention_patterns': ['@Claude'] |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # Mock SSHOUTApiIntegration类 |
| with patch('claude_agent.sshout.SSHOUTApiIntegration') as mock_integration: |
| mock_instance = Mock() |
| mock_integration.return_value = mock_instance |
| |
| result = create_sshout_integration(mock_agent, "test_config") |
| |
| assert result == mock_instance |
| mock_integration.assert_called_once_with(mock_agent, "test_config") |
| |
| def test_create_integration_ssh_mode(self): |
| """测试SSH模式创建集成""" |
| from claude_agent.sshout import create_sshout_integration |
| |
| mock_agent = Mock() |
| |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'connection_mode': 'ssh', |
| 'server': { |
| 'hostname': 'sshout.example.com', |
| 'port': 22, |
| 'username': 'sshout' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/path/to/key' |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| with patch('claude_agent.sshout.SSHOUTIntegration') as mock_integration: |
| mock_instance = Mock() |
| mock_integration.return_value = mock_instance |
| |
| result = create_sshout_integration(mock_agent, "test_config") |
| |
| assert result == mock_instance |
| mock_integration.assert_called_once_with(mock_agent, "test_config") |
| |
| def test_create_integration_invalid_mode(self): |
| """测试无效连接模式""" |
| from claude_agent.sshout import create_sshout_integration |
| |
| mock_agent = Mock() |
| |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'connection_mode': 'invalid_mode' |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| with pytest.raises(ValueError, match="不支持的SSHOUT连接模式"): |
| create_sshout_integration(mock_agent) |
| |
| def test_create_integration_default_api_mode(self): |
| """测试默认API模式""" |
| from claude_agent.sshout import create_sshout_integration |
| |
| mock_agent = Mock() |
| |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| # 不设置connection_mode，应该默认为api |
| mock_config_manager.get_sshout_config.return_value = {} |
| mock_get_config.return_value = mock_config_manager |
| |
| with patch('claude_agent.sshout.SSHOUTApiIntegration') as mock_integration: |
| mock_instance = Mock() |
| mock_integration.return_value = mock_instance |
| |
| result = create_sshout_integration(mock_agent) |
| |
| assert result == mock_instance |
| mock_integration.assert_called_once_with(mock_agent, None) |
| |
| def test_config_error_handling(self): |
| """测试配置错误处理""" |
| from claude_agent.sshout import create_sshout_integration |
| |
| mock_agent = Mock() |
| |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| # 模拟配置获取失败 |
| mock_get_config.side_effect = Exception("配置文件错误") |
| |
| with pytest.raises(Exception): |
| create_sshout_integration(mock_agent) |
| |
| |
| class TestSSHOUTImportStructure: |
| """测试SSHOUT导入结构""" |
| |
| def test_integration_classes_importable(self): |
| """测试集成类可以导入""" |
| try: |
| from claude_agent.sshout.integration import SSHOUTIntegration, SSHOUTConnection, SSHOUTMessage |
| assert SSHOUTIntegration is not None |
| assert SSHOUTConnection is not None |
| assert SSHOUTMessage is not None |
| except ImportError as e: |
| # 如果导入失败，至少确保我们知道原因 |
| assert "SSHOUTIntegration" in str(e) or "SSHOUTConnection" in str(e) or "SSHOUTMessage" in str(e) |
| |
| def test_api_client_classes_importable(self): |
| """测试API客户端类可以导入""" |
| try: |
| from claude_agent.sshout.api_client import SSHOUTApiClient, SSHOUTApiIntegration |
| assert SSHOUTApiClient is not None |
| assert SSHOUTApiIntegration is not None |
| except ImportError as e: |
| # 如果导入失败，至少确保我们知道原因 |
| assert "SSHOUTApiClient" in str(e) or "SSHOUTApiIntegration" in str(e) |
| |
| def test_sshout_module_structure(self): |
| """测试SSHOUT模块结构""" |
| import claude_agent.sshout as sshout |
| |
| # 检查模块是否有__all__属性 |
| if hasattr(sshout, '__all__'): |
| assert isinstance(sshout.__all__, list) |
| assert len(sshout.__all__) > 0 |
| |
| # 检查create_sshout_integration函数 |
| assert hasattr(sshout, 'create_sshout_integration') |
| assert callable(sshout.create_sshout_integration) |
| |
| def test_docstring_exists(self): |
| """测试模块文档字符串存在""" |
| import claude_agent.sshout as sshout |
| |
| assert sshout.__doc__ is not None |
| assert len(sshout.__doc__.strip()) > 0 |
| |
| |
| class TestSSHOUTBasicFunctionality: |
| """测试SSHOUT基础功能""" |
| |
| def test_function_signature(self): |
| """测试create_sshout_integration函数签名""" |
| from claude_agent.sshout import create_sshout_integration |
| import inspect |
| |
| sig = inspect.signature(create_sshout_integration) |
| params = list(sig.parameters.keys()) |
| |
| assert 'agent_core' in params |
| assert 'config_name' in params |
| |
| # 检查config_name是否有默认值 |
| config_param = sig.parameters['config_name'] |
| assert config_param.default is None |
| |
| def test_error_messages(self): |
| """测试错误消息格式""" |
| from claude_agent.sshout import create_sshout_integration |
| |
| mock_agent = Mock() |
| |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'connection_mode': 'unsupported_mode' |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| with pytest.raises(ValueError) as exc_info: |
| create_sshout_integration(mock_agent) |
| |
| error_message = str(exc_info.value) |
| assert "不支持的SSHOUT连接模式" in error_message |
| assert "unsupported_mode" in error_message |
| |
| |
| class TestSSHOUTApiClientBasics: |
| """SSHOUT API Client基础测试""" |
| |
| def test_api_client_initialization(self): |
| """测试API客户端初始化""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key", |
| timeout=10 |
| ) |
| |
| assert client.hostname == "test.example.com" |
| assert client.port == 22 |
| assert client.username == "testuser" |
| assert client.key_path == "/path/to/key" |
| assert client.timeout == 10 |
| assert client.connected is False |
| assert client.client is None |
| assert client.channel is None |
| assert client.message_history == [] |
| assert client.max_history == 100 |
| |
| def test_api_client_initialization_with_mention_patterns(self): |
| """测试API客户端初始化（带mention模式）""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| custom_patterns = ["@TestBot", "TestBot:", "TestBot,"] |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key", |
| mention_patterns=custom_patterns |
| ) |
| |
| assert client.mention_patterns == custom_patterns |
| |
| def test_api_client_default_mention_patterns(self): |
| """测试API客户端默认mention模式""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| expected_patterns = [ |
| "@Claude", "@claude", "@CLAUDE", |
| "Claude:", "claude:", |
| "Claude,", "claude,", |
| "Claude，", "claude，" |
| ] |
| assert client.mention_patterns == expected_patterns |
| |
| def test_sshout_message_dataclass(self): |
| """测试SSHOUT消息数据结构""" |
| from claude_agent.sshout.api_client import SSHOUTMessage, SSHOUTMessageType |
| |
| timestamp = datetime.now() |
| message = SSHOUTMessage( |
| timestamp=timestamp, |
| from_user="user1", |
| to_user="user2", |
| message_type=SSHOUTMessageType.PLAIN, |
| content="Hello world", |
| is_mention=True |
| ) |
| |
| assert message.timestamp == timestamp |
| assert message.from_user == "user1" |
| assert message.to_user == "user2" |
| assert message.message_type == SSHOUTMessageType.PLAIN |
| assert message.content == "Hello world" |
| assert message.is_mention is True |
| |
| def test_sshout_user_dataclass(self): |
| """测试SSHOUT用户数据结构""" |
| from claude_agent.sshout.api_client import SSHOUTUser |
| |
| user = SSHOUTUser( |
| id=123, |
| username="testuser", |
| hostname="client.example.com" |
| ) |
| |
| assert user.id == 123 |
| assert user.username == "testuser" |
| assert user.hostname == "client.example.com" |
| |
| def test_packet_type_enum_values(self): |
| """测试包类型枚举值""" |
| from claude_agent.sshout.api_client import SSHOUTPacketType |
| |
| assert SSHOUTPacketType.HELLO == 1 |
| assert SSHOUTPacketType.GET_ONLINE_USER == 2 |
| assert SSHOUTPacketType.SEND_MESSAGE == 3 |
| assert SSHOUTPacketType.GET_MOTD == 4 |
| assert SSHOUTPacketType.PASS == 128 |
| assert SSHOUTPacketType.ONLINE_USERS_INFO == 129 |
| assert SSHOUTPacketType.RECEIVE_MESSAGE == 130 |
| assert SSHOUTPacketType.USER_STATE_CHANGE == 131 |
| assert SSHOUTPacketType.ERROR == 132 |
| assert SSHOUTPacketType.MOTD == 133 |
| |
| def test_message_type_enum_values(self): |
| """测试消息类型枚举值""" |
| from claude_agent.sshout.api_client import SSHOUTMessageType |
| |
| assert SSHOUTMessageType.PLAIN == 1 |
| assert SSHOUTMessageType.RICH == 2 |
| assert SSHOUTMessageType.IMAGE == 3 |
| |
| def test_error_code_enum_values(self): |
| """测试错误码枚举值""" |
| from claude_agent.sshout.api_client import SSHOUTErrorCode |
| |
| assert SSHOUTErrorCode.SERVER_CLOSED == 1 |
| assert SSHOUTErrorCode.LOCAL_PACKET_CORRUPT == 2 |
| assert SSHOUTErrorCode.LOCAL_PACKET_TOO_LARGE == 3 |
| assert SSHOUTErrorCode.OUT_OF_MEMORY == 4 |
| assert SSHOUTErrorCode.INTERNAL_ERROR == 5 |
| assert SSHOUTErrorCode.USER_NOT_FOUND == 6 |
| assert SSHOUTErrorCode.MOTD_NOT_AVAILABLE == 7 |
| |
| def test_add_message_callback(self): |
| """测试添加消息回调""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| callback = Mock() |
| client.add_message_callback(callback) |
| |
| assert callback in client.message_callbacks |
| |
| def test_add_mention_callback(self): |
| """测试添加mention回调""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| callback = Mock() |
| client.add_mention_callback(callback) |
| |
| assert callback in client.mention_callbacks |
| |
| def test_get_recent_messages_empty(self): |
| """测试获取最近消息（空列表）""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| messages = client.get_recent_messages(5) |
| assert messages == [] |
| |
| def test_get_recent_messages_with_data(self): |
| """测试获取最近消息（有数据）""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient, SSHOUTMessage, SSHOUTMessageType |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 添加一些消息到历史 |
| for i in range(10): |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user=f"user{i}", |
| to_user="public", |
| message_type=SSHOUTMessageType.PLAIN, |
| content=f"Message {i}" |
| ) |
| client.message_history.append(message) |
| |
| # 获取最近5条消息 |
| recent = client.get_recent_messages(5) |
| assert len(recent) == 5 |
| assert recent[0].content == "Message 5" # 最早的 |
| assert recent[4].content == "Message 9" # 最新的 |
| |
| def test_get_recent_messages_limit_exceeds(self): |
| """测试获取最近消息（超出历史数量）""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient, SSHOUTMessage, SSHOUTMessageType |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 只添加3条消息 |
| for i in range(3): |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user=f"user{i}", |
| to_user="public", |
| message_type=SSHOUTMessageType.PLAIN, |
| content=f"Message {i}" |
| ) |
| client.message_history.append(message) |
| |
| # 请求5条消息，应该返回所有3条 |
| recent = client.get_recent_messages(5) |
| assert len(recent) == 3 |
| |
| def test_detect_mention_true(self): |
| """测试mention检测（匹配）""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| test_cases = [ |
| "@Claude hello", |
| "@claude what's up", |
| "Claude: can you help?", |
| "claude, please assist", |
| "Hey Claude， how are you?" |
| ] |
| |
| for content in test_cases: |
| assert client._is_claude_mention(content) is True, f"Failed to detect mention in: {content}" |
| |
| def test_detect_mention_false(self): |
| """测试mention检测（不匹配）""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| test_cases = [ |
| "Hello everyone", |
| "This is a regular message", |
| "Claude3 is nice", # 不是完整的mention |
| "Claudette is here", # 不是Claude |
| "Someone named Claude" # 不是直接mention格式 |
| ] |
| |
| for content in test_cases: |
| assert client._is_claude_mention(content) is False, f"False positive mention in: {content}" |
| |
| def test_detect_mention_custom_patterns(self): |
| """测试自定义mention模式检测""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| custom_patterns = ["@Bot", "Bot:", "Hey Bot"] |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key", |
| mention_patterns=custom_patterns |
| ) |
| |
| assert client._is_claude_mention("@Bot help me") is True |
| assert client._is_claude_mention("Bot: what time is it?") is True |
| assert client._is_claude_mention("Hey Bot can you assist?") is True |
| assert client._is_claude_mention("@Claude hello") is False # 默认模式不再有效 |
| |
| def test_trim_message_history(self): |
| """测试修剪消息历史""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient, SSHOUTMessage, SSHOUTMessageType |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| client.max_history = 5 # 设置小的历史限制 |
| |
| # 添加10条消息 |
| for i in range(10): |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user=f"user{i}", |
| to_user="public", |
| message_type=SSHOUTMessageType.PLAIN, |
| content=f"Message {i}" |
| ) |
| client.message_history.append(message) |
| |
| # 手动触发历史修剪 (实际实现中是自动的) |
| if len(client.message_history) > client.max_history: |
| client.message_history = client.message_history[-client.max_history:] |
| |
| # 应该只保留最近的5条消息 |
| assert len(client.message_history) == 5 |
| assert client.message_history[0].content == "Message 5" |
| assert client.message_history[4].content == "Message 9" |
| |
| def test_connection_status(self): |
| """测试连接状态检查""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 测试未连接状态 |
| status = client.get_connection_status() |
| assert status['connected'] is False |
| assert status['server'] is None |
| assert status['message_count'] == 0 |
| |
| # 测试已连接状态 |
| client.connected = True |
| client.my_user_id = 123 |
| client.my_username = "testuser" |
| |
| status = client.get_connection_status() |
| assert status['connected'] is True |
| assert status['server'] == "test.example.com:22" |
| assert status['my_user_id'] == 123 |
| assert status['my_username'] == "testuser" |
| |
| |
| class TestSSHOUTApiClientMessageMethods: |
| """SSHOUT API Client消息方法测试""" |
| |
| @pytest.mark.asyncio |
| async def test_send_message_not_connected(self): |
| """测试未连接时发送消息""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| result = await client.send_message("public", "Hello world") |
| assert result is False |
| |
| @pytest.mark.asyncio |
| async def test_send_global_message_not_connected(self): |
| """测试未连接时发送全局消息""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| result = await client.send_global_message("Hello everyone") |
| assert result is False |
| |
| def test_get_context_messages(self): |
| """测试获取上下文消息""" |
| from claude_agent.sshout.api_client import SSHOUTApiClient, SSHOUTMessage, SSHOUTMessageType |
| |
| client = SSHOUTApiClient( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 添加一些历史消息 |
| base_time = datetime.now() |
| for i in range(10): |
| message_time = base_time.replace(second=i) |
| message = SSHOUTMessage( |
| timestamp=message_time, |
| from_user=f"user{i}", |
| to_user="public", |
| message_type=SSHOUTMessageType.PLAIN, |
| content=f"Message {i}" |
| ) |
| client.message_history.append(message) |
| |
| # 获取第5条消息之前的上下文 |
| target_time = base_time.replace(second=5) |
| context = client.get_context_messages(target_time, count=3) |
| |
| assert len(context) == 3 |
| assert context[0].content == "Message 2" # 最早的上下文 |
| assert context[2].content == "Message 4" # 最近的上下文（时间小于target_time） |
| |
| |
| class TestSSHOUTApiIntegrationBasics: |
| """SSHOUT API Integration基础测试""" |
| |
| @pytest.fixture |
| def mock_agent(self): |
| """创建模拟Agent""" |
| return Mock() |
| |
| def test_integration_initialization_missing_config(self, mock_agent): |
| """测试集成初始化（缺少配置）""" |
| from claude_agent.sshout.api_client import SSHOUTApiIntegration |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = {} # 空配置 |
| mock_get_config.return_value = mock_config_manager |
| |
| with pytest.raises(ValueError, match="SSHOUT配置缺少必需的段落"): |
| SSHOUTApiIntegration(mock_agent) |
| |
| def test_integration_initialization_missing_key_file(self, mock_agent): |
| """测试集成初始化（私钥文件不存在）""" |
| from claude_agent.sshout.api_client import SSHOUTApiIntegration |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/nonexistent/key' |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| with pytest.raises(FileNotFoundError, match="SSH私钥文件不存在"): |
| SSHOUTApiIntegration(mock_agent) |
| |
| @pytest.mark.asyncio |
| async def test_disconnect_from_sshout_api(self, mock_agent): |
| """测试断开SSHOUT API连接""" |
| from claude_agent.sshout.api_client import SSHOUTApiIntegration |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTApiIntegration(mock_agent) |
| |
| # 设置模拟客户端 |
| mock_client = AsyncMock() |
| integration.client = mock_client |
| |
| await integration.disconnect_from_sshout_api() |
| |
| mock_client.disconnect.assert_called_once() |
| assert integration.client is None |
| |
| def test_get_connection_status_no_client(self, mock_agent): |
| """测试获取连接状态（无客户端）""" |
| from claude_agent.sshout.api_client import SSHOUTApiIntegration |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTApiIntegration(mock_agent) |
| status = integration.get_connection_status() |
| |
| assert status['connected'] is False |
| assert status['api_version'] == '1.0' |
| |
| @pytest.mark.asyncio |
| async def test_send_message_no_client(self, mock_agent): |
| """测试发送消息（无客户端）""" |
| from claude_agent.sshout.api_client import SSHOUTApiIntegration |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTApiIntegration(mock_agent) |
| result = await integration.send_message("Test message") |
| |
| assert result is False |
| |
| def test_clean_response_for_sshout(self, mock_agent): |
| """测试清理响应文本""" |
| from claude_agent.sshout.api_client import SSHOUTApiIntegration |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_config_manager.get.return_value = 0 # 无长度限制 |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTApiIntegration(mock_agent) |
| |
| # 测试清理markdown格式 |
| text = "这是**粗体**和*斜体*以及`代码`文本" |
| cleaned = integration._clean_response_for_sshout(text) |
| assert cleaned == "这是粗体和斜体以及代码文本" |
| |
| # 测试清理多余换行 |
| text = "行1\n\n\n\n行2" |
| cleaned = integration._clean_response_for_sshout(text) |
| assert cleaned == "行1\n\n行2" |
| |
| |
| class TestSSHOUTConnectionBasics: |
| """SSHOUT Connection基础测试""" |
| |
| def test_connection_initialization(self): |
| """测试连接初始化""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| assert connection.hostname == "test.example.com" |
| assert connection.port == 22 |
| assert connection.username == "testuser" |
| assert connection.key_path == "/path/to/key" |
| assert connection.connected is False |
| assert connection.client is None |
| assert connection.shell is None |
| assert len(connection.message_callbacks) == 0 |
| assert len(connection.mention_callbacks) == 0 |
| assert len(connection.message_history) == 0 |
| assert connection.max_history == 100 |
| |
| def test_connection_initialization_with_custom_patterns(self): |
| """测试带自定义mention模式的连接初始化""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| custom_patterns = ["@Bot", "Bot:", "Hey Bot"] |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key", |
| mention_patterns=custom_patterns |
| ) |
| |
| assert connection.mention_patterns == custom_patterns |
| |
| def test_connection_default_mention_patterns(self): |
| """测试默认mention模式""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| expected_patterns = [ |
| "@Claude", "@claude", "@CLAUDE", |
| "Claude:", "claude:", |
| "Claude,", "claude,", |
| "Claude，", "claude，" |
| ] |
| assert connection.mention_patterns == expected_patterns |
| |
| def test_add_message_callback(self): |
| """测试添加消息回调""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| callback = Mock() |
| connection.add_message_callback(callback) |
| |
| assert callback in connection.message_callbacks |
| |
| def test_add_mention_callback(self): |
| """测试添加mention回调""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| callback = Mock() |
| connection.add_mention_callback(callback) |
| |
| assert callback in connection.mention_callbacks |
| |
| def test_get_recent_messages_empty(self): |
| """测试获取最近消息（空列表）""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| messages = connection.get_recent_messages(5) |
| assert messages == [] |
| |
| def test_get_recent_messages_with_data(self): |
| """测试获取最近消息（有数据）""" |
| from claude_agent.sshout.integration import SSHOUTConnection, SSHOUTMessage |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 添加一些消息到历史 |
| for i in range(10): |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| username=f"user{i}", |
| content=f"Message {i}" |
| ) |
| connection.message_history.append(message) |
| |
| # 获取最近5条消息 |
| recent = connection.get_recent_messages(5) |
| assert len(recent) == 5 |
| assert recent[0].content == "Message 5" # 最早的 |
| assert recent[4].content == "Message 9" # 最新的 |
| |
| def test_get_context_messages(self): |
| """测试获取上下文消息""" |
| from claude_agent.sshout.integration import SSHOUTConnection, SSHOUTMessage |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 添加一些历史消息 |
| base_time = datetime.now() |
| for i in range(10): |
| message_time = base_time.replace(second=i) |
| message = SSHOUTMessage( |
| timestamp=message_time, |
| username=f"user{i}", |
| content=f"Message {i}" |
| ) |
| connection.message_history.append(message) |
| |
| # 获取第5条消息之前的上下文 |
| target_time = base_time.replace(second=5) |
| context = connection.get_context_messages(target_time, count=3) |
| |
| assert len(context) == 3 |
| assert context[0].content == "Message 2" # 最早的上下文 |
| assert context[2].content == "Message 4" # 最近的上下文（时间小于target_time） |
| |
| def test_is_claude_mention_detection(self): |
| """测试@Claude提及检测""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 测试匹配的情况 |
| test_cases_true = [ |
| "@Claude hello", |
| "@claude what's up", |
| "Claude: can you help?", |
| "claude, please assist", |
| "Hey Claude， how are you?" |
| ] |
| |
| for content in test_cases_true: |
| assert connection._is_claude_mention(content) is True, f"Failed to detect mention in: {content}" |
| |
| # 测试不匹配的情况 |
| test_cases_false = [ |
| "Hello everyone", |
| "This is a regular message", |
| "Claude3 is nice", # 不是完整的mention |
| "Claudette is here", # 不是Claude |
| "Someone named Claude" # 不是直接mention格式 |
| ] |
| |
| for content in test_cases_false: |
| assert connection._is_claude_mention(content) is False, f"False positive mention in: {content}" |
| |
| def test_is_claude_mention_custom_patterns(self): |
| """测试自定义mention模式检测""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| custom_patterns = ["@Bot", "Bot:", "Hey Bot"] |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key", |
| mention_patterns=custom_patterns |
| ) |
| |
| assert connection._is_claude_mention("@Bot help me") is True |
| assert connection._is_claude_mention("Bot: what time is it?") is True |
| assert connection._is_claude_mention("Hey Bot can you assist?") is True |
| assert connection._is_claude_mention("@Claude hello") is False # 默认模式不再有效 |
| |
| @pytest.mark.asyncio |
| async def test_send_message_not_connected(self): |
| """测试未连接时发送消息""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| result = await connection.send_message("Hello world") |
| assert result is False |
| |
| @pytest.mark.asyncio |
| async def test_disconnect_cleanup(self): |
| """测试断开连接时清理资源""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 设置已连接状态 |
| connection.connected = True |
| connection.shell = Mock() |
| connection.client = Mock() |
| |
| # 执行断开连接 |
| await connection.disconnect() |
| |
| # 验证状态重置 |
| assert connection.connected is False |
| assert connection.shell is None |
| assert connection.client is None |
| |
| |
| class TestSSHOUTMessageParsing: |
| """SSHOUT消息解析测试""" |
| |
| def test_sshout_message_creation(self): |
| """测试SSHOUT消息创建""" |
| from claude_agent.sshout.integration import SSHOUTMessage |
| |
| timestamp = datetime.now() |
| message = SSHOUTMessage( |
| timestamp=timestamp, |
| username="testuser", |
| content="Hello world", |
| is_mention=True |
| ) |
| |
| assert message.timestamp == timestamp |
| assert message.username == "testuser" |
| assert message.content == "Hello world" |
| assert message.is_mention is True |
| |
| def test_clean_ansi_codes(self): |
| """测试清理ANSI转义码""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 测试ANSI转义码清理 |
| test_cases = [ |
| ("\x1b[1;34mHello\x1b[0m", "Hello"), |
| ("[1;34mWorld[0m", "World"), |
| ("Normal text", "Normal text"), |
| ("\x1b[31mRed\x1b[0m and \x1b[32mGreen\x1b[0m", "Red and Green"), |
| ] |
| |
| for input_text, expected in test_cases: |
| result = connection._clean_ansi_codes(input_text) |
| assert result == expected, f"Expected '{expected}', got '{result}' for input '{input_text}'" |
| |
| def test_parse_message_various_formats(self): |
| """测试解析各种消息格式""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| test_cases = [ |
| # 格式1: [HH:MM:SS] <username> message |
| "[12:34:56] <alice> Hello everyone", |
| # 格式2: <username> message |
| "<bob> How are you?", |
| # 格式3: username: message |
| "charlie: Good morning!", |
| # 格式4: [HH:MM:SS] username: message |
| "[09:15:30] dave: Testing message", |
| ] |
| |
| for line in test_cases: |
| message = connection._parse_message(line) |
| assert message is not None, f"Failed to parse: {line}" |
| assert message.username is not None |
| assert message.content is not None |
| assert message.timestamp is not None |
| |
| def test_parse_message_with_ansi_codes(self): |
| """测试解析带ANSI转义码的消息""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| # 带ANSI转义码的消息 |
| line = "\x1b[1;34m[12:34:56] <alice> \x1b[0mHello \x1b[31mworld\x1b[0m" |
| message = connection._parse_message(line) |
| |
| assert message is not None |
| assert message.username == "alice" |
| assert message.content == "Hello world" |
| |
| def test_parse_message_invalid_format(self): |
| """测试解析无效格式的消息""" |
| from claude_agent.sshout.integration import SSHOUTConnection |
| |
| connection = SSHOUTConnection( |
| hostname="test.example.com", |
| port=22, |
| username="testuser", |
| key_path="/path/to/key" |
| ) |
| |
| invalid_lines = [ |
| "Just some random text", |
| ">>> System message <<<", |
| "", |
| " ", |
| "12:34:56 without brackets or username", |
| ] |
| |
| for line in invalid_lines: |
| message = connection._parse_message(line) |
| # Some of these might return None, which is expected |
| if message is not None: |
| # If parsed, should have valid fields |
| assert message.username is not None |
| assert message.content is not None |
| |
| |
| class TestSSHOUTIntegrationAdvanced: |
| """SSHOUT Integration高级测试""" |
| |
| @pytest.fixture |
| def mock_agent(self): |
| """创建模拟Agent""" |
| return Mock() |
| |
| def test_integration_initialization_with_valid_config(self, mock_agent): |
| """测试集成初始化（有效配置）""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/path/to/key' |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| |
| assert integration.agent == mock_agent |
| assert integration.connection is None |
| assert integration.sshout_config is not None |
| |
| def test_integration_initialization_missing_server_config(self, mock_agent): |
| """测试集成初始化（缺少服务器配置键）""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': { |
| 'hostname': 'test.example.com' |
| # 缺少 port 和 username |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/path/to/key' |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| with pytest.raises(ValueError, match="SSHOUT服务器配置缺少必需的键"): |
| SSHOUTIntegration(mock_agent) |
| |
| def test_integration_initialization_missing_ssh_key(self, mock_agent): |
| """测试集成初始化（缺少SSH密钥配置）""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| # 缺少 private_key_path |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| with pytest.raises(ValueError, match="SSHOUT配置缺少SSH私钥路径"): |
| SSHOUTIntegration(mock_agent) |
| |
| @pytest.mark.asyncio |
| async def test_disconnect_from_sshout(self, mock_agent): |
| """测试断开SSHOUT连接""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| |
| # 设置模拟连接 |
| mock_connection = AsyncMock() |
| integration.connection = mock_connection |
| |
| await integration.disconnect_from_sshout() |
| |
| mock_connection.disconnect.assert_called_once() |
| assert integration.connection is None |
| |
| def test_on_message_received(self, mock_agent): |
| """测试接收普通消息回调""" |
| from claude_agent.sshout.integration import SSHOUTIntegration, SSHOUTMessage |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| |
| # 创建测试消息 |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| username="testuser", |
| content="Hello world" |
| ) |
| |
| # 调用回调函数不应该抛出异常 |
| integration._on_message_received(message) |
| |
| def test_get_connection_status_no_connection(self, mock_agent): |
| """测试获取连接状态（无连接）""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| status = integration.get_connection_status() |
| |
| assert status['connected'] is False |
| assert status['server'] is None |
| assert status['message_count'] == 0 |
| |
| def test_get_connection_status_with_connection(self, mock_agent): |
| """测试获取连接状态（有连接）""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| |
| # 设置模拟连接 |
| mock_connection = Mock() |
| mock_connection.connected = True |
| mock_connection.hostname = "test.com" |
| mock_connection.port = 22 |
| mock_connection.message_history = ["msg1", "msg2"] |
| mock_connection.get_recent_messages.return_value = [] |
| integration.connection = mock_connection |
| |
| status = integration.get_connection_status() |
| |
| assert status['connected'] is True |
| assert status['server'] == "test.com:22" |
| assert status['message_count'] == 2 |
| |
| @pytest.mark.asyncio |
| async def test_send_message_no_connection(self, mock_agent): |
| """测试发送消息（无连接）""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| result = await integration.send_message("Test message") |
| |
| assert result is False |
| |
| @pytest.mark.asyncio |
| async def test_send_message_with_connection(self, mock_agent): |
| """测试发送消息（有连接）""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| |
| # 设置模拟连接 |
| mock_connection = AsyncMock() |
| mock_connection.connected = True |
| mock_connection.send_message.return_value = True |
| integration.connection = mock_connection |
| |
| result = await integration.send_message("Test message") |
| |
| assert result is True |
| mock_connection.send_message.assert_called_once_with("Test message") |
| |
| def test_clean_response_for_sshout_formatting(self, mock_agent): |
| """测试清理响应文本格式""" |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists', return_value=True): |
| |
| mock_config_manager = Mock() |
| mock_config_manager.get_sshout_config.return_value = { |
| 'server': {'hostname': 'test.com', 'port': 22, 'username': 'user'}, |
| 'ssh_key': {'private_key_path': '/test/key'} |
| } |
| mock_config_manager.get.return_value = 0 # 无长度限制 |
| mock_get_config.return_value = mock_config_manager |
| |
| integration = SSHOUTIntegration(mock_agent) |
| |
| # 测试清理markdown格式 |
| text = "这是**粗体**和*斜体*以及`代码`文本" |
| cleaned = integration._clean_response_for_sshout(text) |
| assert cleaned == "这是粗体和斜体以及代码文本" |
| |
| # 测试清理多余换行 |
| text = "行1\n\n\n\n行2" |
| cleaned = integration._clean_response_for_sshout(text) |
| assert cleaned == "行1\n\n行2" |
| |
| # 测试长度限制 |
| mock_config_manager.get.return_value = 10 # 设置长度限制 |
| long_text = "这是一个很长的文本内容用于测试长度限制功能" |
| cleaned = integration._clean_response_for_sshout(long_text) |
| assert cleaned == "这是一个很长的文本内..." |
| assert len(cleaned) <= 13 # 10 + "..." |