| """ |
| SSHOUT用户回复功能的单元测试 |
| 测试@Claude提及处理和回复格式 |
| """ |
| |
| import pytest |
| import asyncio |
| from unittest.mock import Mock, AsyncMock, patch |
| from datetime import datetime |
| |
| import sys |
| import os |
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../../src')) |
| |
| from claude_agent.sshout.api_client import SSHOUTApiIntegration, SSHOUTMessage, SSHOUTMessageType |
| from claude_agent.sshout.integration import SSHOUTIntegration |
| |
| |
| class TestSSHOUTUserReplyFunctionality: |
| """测试SSHOUT用户回复功能""" |
| |
| def setup_method(self): |
| """设置测试""" |
| self.mock_agent = Mock() |
| self.mock_agent.process_user_input = AsyncMock(return_value="Hello! How can I help you?") |
| |
| @pytest.mark.asyncio |
| async def test_api_client_reply_without_at_prefix(self): |
| """测试API客户端回复不包含@前缀""" |
| # 创建mock配置管理器 |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'api', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 200 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 200 |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTApiIntegration(self.mock_agent) |
| |
| # 创建mock客户端 |
| mock_client = Mock() |
| mock_client.send_global_message = AsyncMock(return_value=True) |
| mock_client.get_context_messages.return_value = [] |
| integration.client = mock_client |
| |
| # 创建测试消息 |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user="testuser", |
| to_user="GLOBAL", |
| message_type=SSHOUTMessageType.PLAIN, |
| content="@Claude help me with something" |
| ) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 验证回复没有@前缀 |
| mock_client.send_global_message.assert_called_once() |
| call_args = mock_client.send_global_message.call_args[0][0] |
| |
| # 确保回复不以@testuser开头 |
| assert not call_args.startswith("@testuser"), f"回复包含@前缀: {call_args}" |
| assert call_args == "Hello! How can I help you?", f"回复内容不正确: {call_args}" |
| |
| @pytest.mark.asyncio |
| async def test_ssh_integration_reply_without_at_prefix(self): |
| """测试SSH集成回复不包含@前缀""" |
| # 创建mock配置管理器 |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'ssh', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 200 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 200 |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTIntegration(self.mock_agent) |
| |
| # 创建mock连接 |
| mock_connection = Mock() |
| mock_connection.send_message = AsyncMock(return_value=True) |
| integration.connection = mock_connection |
| |
| # 创建测试消息 |
| message = Mock() |
| message.username = "testuser" |
| message.content = "@Claude what's the weather like?" |
| message.timestamp = datetime.now() |
| |
| # Mock get_context_messages方法 |
| integration.connection.get_context_messages = Mock(return_value=[]) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 验证回复没有@前缀 |
| mock_connection.send_message.assert_called_once() |
| call_args = mock_connection.send_message.call_args[0][0] |
| |
| # 确保回复不以@testuser开头 |
| assert not call_args.startswith("@testuser"), f"回复包含@前缀: {call_args}" |
| assert call_args == "Hello! How can I help you?", f"回复内容不正确: {call_args}" |
| |
| @pytest.mark.asyncio |
| async def test_reply_content_cleaning(self): |
| """测试回复内容清理功能""" |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'api', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 50 # 设置较短的长度限制 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 50 |
| |
| # Mock agent返回带格式的响应 |
| self.mock_agent.process_user_input = AsyncMock( |
| return_value="**Bold text** and *italic text* with `code` and multiple\n\nlines of text" |
| ) |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTApiIntegration(self.mock_agent) |
| |
| # 创建mock客户端 |
| mock_client = Mock() |
| mock_client.send_global_message = AsyncMock(return_value=True) |
| mock_client.get_context_messages.return_value = [] |
| integration.client = mock_client |
| |
| # 创建测试消息 |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user="testuser", |
| to_user="GLOBAL", |
| message_type=SSHOUTMessageType.PLAIN, |
| content="@Claude explain something complex" |
| ) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 验证回复内容已清理且限制长度 |
| mock_client.send_global_message.assert_called_once() |
| call_args = mock_client.send_global_message.call_args[0][0] |
| |
| # 验证格式已清理 |
| assert "**" not in call_args, "粗体格式未清理" |
| assert "*" not in call_args, "斜体格式未清理" |
| assert "`" not in call_args, "代码格式未清理" |
| # 换行符现在应该被保留 |
| # 注意：由于有长度限制(50字符)，可能会被截断 |
| |
| # 验证长度限制 |
| assert len(call_args) <= 53, f"回复长度超限: {len(call_args)} > 53" # 50 + "..." |
| |
| # 验证没有@前缀 |
| assert not call_args.startswith("@testuser"), f"回复包含@前缀: {call_args}" |
| |
| @pytest.mark.asyncio |
| async def test_multiple_users_reply_format(self): |
| """测试多用户场景下的回复格式""" |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'api', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 200 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 200 |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTApiIntegration(self.mock_agent) |
| |
| # 创建mock客户端 |
| mock_client = Mock() |
| mock_client.send_global_message = AsyncMock(return_value=True) |
| mock_client.get_context_messages.return_value = [] |
| integration.client = mock_client |
| |
| # 测试不同用户名 |
| test_users = ["alice", "bob", "user123", "special-user", "用户名"] |
| |
| for username in test_users: |
| # 重置mock |
| mock_client.send_global_message.reset_mock() |
| |
| # 创建测试消息 |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user=username, |
| to_user="GLOBAL", |
| message_type=SSHOUTMessageType.PLAIN, |
| content=f"@Claude hello from {username}" |
| ) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 验证回复格式 |
| mock_client.send_global_message.assert_called_once() |
| call_args = mock_client.send_global_message.call_args[0][0] |
| |
| # 确保没有@用户名前缀 |
| assert not call_args.startswith(f"@{username}"), f"用户{username}的回复包含@前缀: {call_args}" |
| assert call_args == "Hello! How can I help you?", f"用户{username}的回复内容不正确: {call_args}" |
| |
| @pytest.mark.asyncio |
| async def test_reply_failure_handling(self): |
| """测试回复失败处理""" |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'api', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 200 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 200 |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTApiIntegration(self.mock_agent) |
| integration.logger = Mock() |
| |
| # 创建mock客户端，发送失败 |
| mock_client = Mock() |
| mock_client.send_global_message = AsyncMock(return_value=False) |
| mock_client.get_context_messages.return_value = [] |
| integration.client = mock_client |
| |
| # 创建测试消息 |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user="testuser", |
| to_user="GLOBAL", |
| message_type=SSHOUTMessageType.PLAIN, |
| content="@Claude test message" |
| ) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 验证错误日志 |
| integration.logger.error.assert_called_with("❌ 回复失败") |
| |
| @pytest.mark.asyncio |
| async def test_agent_processing_error_handling(self): |
| """测试Agent处理错误的情况""" |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'api', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 200 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 200 |
| |
| # Mock agent抛出异常 |
| self.mock_agent.process_user_input = AsyncMock(side_effect=Exception("Agent processing error")) |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTApiIntegration(self.mock_agent) |
| integration.logger = Mock() |
| |
| # 创建mock客户端 |
| mock_client = Mock() |
| mock_client.send_global_message = AsyncMock(return_value=True) |
| mock_client.get_context_messages.return_value = [] |
| integration.client = mock_client |
| |
| # 创建测试消息 |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user="testuser", |
| to_user="GLOBAL", |
| message_type=SSHOUTMessageType.PLAIN, |
| content="@Claude cause an error" |
| ) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 验证错误处理 |
| integration.logger.error.assert_called() |
| error_call_args = integration.logger.error.call_args[0][0] |
| assert "处理@Claude提及时出错" in error_call_args |
| |
| # 验证没有发送回复 |
| mock_client.send_global_message.assert_not_called() |
| |
| |
| class TestSSHOUTReplyFormatRegression: |
| """回归测试：确保@前缀不会再次出现""" |
| |
| @pytest.mark.asyncio |
| async def test_api_client_no_at_prefix_regression(self): |
| """回归测试：API客户端回复不包含@前缀""" |
| # 这是一个专门的回归测试，确保之前修复的@前缀问题不会重现 |
| mock_agent = Mock() |
| mock_agent.process_user_input = AsyncMock(return_value="Simple response") |
| |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'api', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 200 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 200 |
| |
| with patch('claude_agent.sshout.api_client.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.api_client.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTApiIntegration(mock_agent) |
| |
| # 创建mock客户端 |
| mock_client = Mock() |
| mock_client.send_global_message = AsyncMock(return_value=True) |
| mock_client.get_context_messages.return_value = [] |
| integration.client = mock_client |
| |
| # 创建测试消息 |
| message = SSHOUTMessage( |
| timestamp=datetime.now(), |
| from_user="regressiontestuser", |
| to_user="GLOBAL", |
| message_type=SSHOUTMessageType.PLAIN, |
| content="@Claude this is a regression test" |
| ) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 严格验证回复格式 |
| mock_client.send_global_message.assert_called_once() |
| actual_reply = mock_client.send_global_message.call_args[0][0] |
| |
| # 多重检查确保没有@前缀 |
| assert not actual_reply.startswith("@"), f"REGRESSION: 回复包含@前缀: '{actual_reply}'" |
| assert not actual_reply.startswith("@regressiontestuser"), f"REGRESSION: 回复包含@用户名前缀: '{actual_reply}'" |
| assert "@regressiontestuser" not in actual_reply, f"REGRESSION: 回复包含@用户名: '{actual_reply}'" |
| assert actual_reply == "Simple response", f"REGRESSION: 回复内容不正确: '{actual_reply}'" |
| |
| @pytest.mark.asyncio |
| async def test_ssh_integration_no_at_prefix_regression(self): |
| """回归测试：SSH集成回复不包含@前缀""" |
| mock_agent = Mock() |
| mock_agent.process_user_input = AsyncMock(return_value="SSH response") |
| |
| mock_config_manager = Mock() |
| mock_config = { |
| 'connection_mode': 'ssh', |
| 'mention_patterns': ['@Claude'], |
| 'server': { |
| 'hostname': 'test.example.com', |
| 'port': 22333, |
| 'username': 'testuser' |
| }, |
| 'ssh_key': { |
| 'private_key_path': '/fake/key/path', |
| 'timeout': 10 |
| }, |
| 'message': { |
| 'max_history': 100, |
| 'context_count': 5, |
| 'max_reply_length': 200 |
| } |
| } |
| mock_config_manager.get_sshout_config.return_value = mock_config |
| mock_config_manager.get.return_value = 200 |
| |
| with patch('claude_agent.sshout.integration.get_config_manager') as mock_get_config, \ |
| patch('claude_agent.sshout.integration.os.path.exists') as mock_exists: |
| |
| mock_get_config.return_value = mock_config_manager |
| mock_exists.return_value = True |
| |
| integration = SSHOUTIntegration(mock_agent) |
| |
| # 创建mock连接 |
| mock_connection = Mock() |
| mock_connection.send_message = AsyncMock(return_value=True) |
| integration.connection = mock_connection |
| |
| # 创建测试消息 |
| message = Mock() |
| message.username = "sshuser" |
| message.content = "@Claude SSH regression test" |
| message.timestamp = datetime.now() |
| |
| # Mock get_context_messages方法 |
| integration.connection.get_context_messages = Mock(return_value=[]) |
| |
| # 处理@Claude提及 |
| await integration._on_claude_mentioned(message) |
| |
| # 严格验证回复格式 |
| mock_connection.send_message.assert_called_once() |
| actual_reply = mock_connection.send_message.call_args[0][0] |
| |
| # 多重检查确保没有@前缀 |
| assert not actual_reply.startswith("@"), f"SSH REGRESSION: 回复包含@前缀: '{actual_reply}'" |
| assert not actual_reply.startswith("@sshuser"), f"SSH REGRESSION: 回复包含@用户名前缀: '{actual_reply}'" |
| assert "@sshuser" not in actual_reply, f"SSH REGRESSION: 回复包含@用户名: '{actual_reply}'" |
| assert actual_reply == "SSH response", f"SSH REGRESSION: 回复内容不正确: '{actual_reply}'" |
| |
| |
| if __name__ == '__main__': |
| pytest.main([__file__]) |