tests/integration/test_sshout_real_connection.py

"""
SSHOUT实际连接集成测试
测试真实的SSH连接和消息处理功能
"""

import unittest
import asyncio
import os
import sys
from unittest.mock import patch, AsyncMock

# 添加项目根目录到Python路径
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.join(project_root, 'src'))

from claude_agent.sshout.integration import SSHOUTIntegration, SSHOUTConnection
from claude_agent.utils.config import get_config_manager


class TestSSHOUTRealConnection(unittest.IsolatedAsyncioTestCase):
    """SSHOUT实际连接测试"""

    def setUp(self):
        """设置测试环境"""
        self.mock_agent = AsyncMock()
        # 确保使用默认配置
        self.integration = SSHOUTIntegration(self.mock_agent, config_name="default")

    def test_config_loading(self):
        """测试配置加载"""
        config = self.integration.config_manager.get_sshout_config()

        # 验证配置结构
        self.assertIn('server', config)
        self.assertIn('ssh_key', config)
        self.assertIn('message', config)

        # 验证服务器配置
        server_config = config['server']
        self.assertEqual(server_config['hostname'], 'tianjin1.rivoreo.one')
        self.assertEqual(server_config['port'], 22333)
        self.assertEqual(server_config['username'], 'sshout')

        # 验证SSH密钥路径
        ssh_config = config['ssh_key']
        self.assertIn('private_key_path', ssh_config)
        key_path = ssh_config['private_key_path']
        self.assertTrue(os.path.exists(key_path), f"SSH密钥文件不存在: {key_path}")

    def test_config_validation(self):
        """测试配置验证"""
        # 正常情况应该不抛出异常
        try:
            self.integration._validate_config()
        except Exception as e:
            self.fail(f"配置验证失败: {e}")

    @patch('claude_agent.sshout.integration.SSHOUTConnection')
    async def test_connection_attempt(self, mock_connection_class):
        """测试连接尝试"""
        # 设置mock连接
        mock_connection = AsyncMock()
        mock_connection.connect.return_value = True
        mock_connection_class.return_value = mock_connection

        # 尝试连接
        success = await self.integration.connect_to_sshout()

        # 验证结果
        self.assertTrue(success)
        mock_connection_class.assert_called_once()
        mock_connection.connect.assert_called_once()

        # 验证连接参数
        call_args = mock_connection_class.call_args
        self.assertEqual(call_args[1]['hostname'], 'tianjin1.rivoreo.one')
        self.assertEqual(call_args[1]['port'], 22333)
        self.assertEqual(call_args[1]['username'], 'sshout')

    async def test_mention_detection_with_config(self):
        """测试使用配置的@Claude检测"""
        # 创建测试连接
        connection = SSHOUTConnection("test", 22, "test", "test")

        # 获取配置的提及模式
        mention_patterns = self.integration.config_manager.get('sshout.mention_patterns', [])
        self.assertGreater(len(mention_patterns), 0)

        # 测试配置中的每个模式
        for pattern in mention_patterns[:3]:  # 只测试前3个模式
            with self.subTest(pattern=pattern):
                # 构造测试消息
                test_message = f"Hello {pattern} how are you?"
                result = connection._is_claude_mention(test_message)
                self.assertTrue(result, f"应该检测到模式: {pattern}")

    async def test_message_length_limit_from_config(self):
        """测试从配置获取消息长度限制"""
        max_length = self.integration.config_manager.get('sshout.message.max_reply_length', 200)

        # 创建超长回复
        long_response = "这是一个很长的回复。" * 50

        # 清理回复
        cleaned = self.integration._clean_response_for_sshout(long_response)

        # 验证长度限制
        self.assertLessEqual(len(cleaned), max_length + 3)  # +3 for "..."

    def test_ssh_key_permissions(self):
        """测试SSH密钥文件权限"""
        config = self.integration.config_manager.get_sshout_config()
        key_path = config['ssh_key']['private_key_path']

        if os.path.exists(key_path):
            # 检查文件权限 (应该是只有所有者可读)
            stat_info = os.stat(key_path)
            permissions = stat_info.st_mode & 0o777

            # 私钥文件应该是 600 (只有所有者可读写) 或更严格
            self.assertLessEqual(permissions, 0o600,
                               f"SSH私钥文件权限过于宽松: {oct(permissions)}")


class TestSSHOUTConnectionReal(unittest.IsolatedAsyncioTestCase):
    """SSHOUT连接真实性测试（需要实际网络）"""

    def setUp(self):
        """设置测试环境"""
        self.config_manager = get_config_manager("default")
        self.sshout_config = self.config_manager.get_sshout_config()

    @unittest.skipUnless(os.getenv('TEST_REAL_SSHOUT') == '1',
                         "需要设置 TEST_REAL_SSHOUT=1 环境变量来运行实际连接测试")
    async def test_real_ssh_connection(self):
        """测试实际的SSH连接（可选测试）"""
        server_config = self.sshout_config['server']
        ssh_config = self.sshout_config['ssh_key']

        connection = SSHOUTConnection(
            hostname=server_config['hostname'],
            port=server_config['port'],
            username=server_config['username'],
            key_path=ssh_config['private_key_path']
        )

        try:
            # 尝试连接（有超时保护）
            success = await asyncio.wait_for(
                connection.connect(),
                timeout=ssh_config.get('timeout', 10)
            )

            if success:
                print("✅ SSHOUT实际连接测试成功")
                # 发送测试消息
                test_message = f"[测试] Claude Agent 连接测试 - {asyncio.get_event_loop().time()}"
                send_success = await connection.send_message(test_message)

                if send_success:
                    print("✅ SSHOUT消息发送测试成功")
                else:
                    print("❌ SSHOUT消息发送失败")

                # 清理连接
                await connection.disconnect()
            else:
                print("❌ SSHOUT实际连接失败")

        except asyncio.TimeoutError:
            print("⏰ SSHOUT连接超时")
            self.skipTest("SSHOUT连接超时，跳过实际连接测试")

        except Exception as e:
            print(f"💥 SSHOUT连接异常: {e}")
            # 不将异常视为测试失败，因为可能是网络或服务器问题
            self.skipTest(f"SSHOUT连接异常，跳过测试: {e}")

    async def test_network_connectivity(self):
        """测试网络连通性"""
        import socket

        server_config = self.sshout_config['server']
        hostname = server_config['hostname']
        port = server_config['port']

        try:
            # 创建socket连接测试
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)

            result = sock.connect_ex((hostname, port))
            sock.close()

            if result == 0:
                print(f"✅ 网络连通性测试通过: {hostname}:{port}")
            else:
                print(f"❌ 网络连通性测试失败: {hostname}:{port}")
                self.skipTest(f"无法连接到 {hostname}:{port}")

        except Exception as e:
            print(f"💥 网络连通性测试异常: {e}")
            self.skipTest(f"网络连通性测试异常: {e}")

    def test_config_completeness(self):
        """测试配置完整性"""
        # 验证所有必需的配置项
        required_paths = [
            'server.hostname',
            'server.port',
            'server.username',
            'ssh_key.private_key_path',
            'message.max_history',
            'message.context_count',
            'message.max_reply_length',
            'mention_patterns'
        ]

        for path in required_paths:
            with self.subTest(config_path=path):
                value = self.config_manager.get(f'sshout.{path}')
                self.assertIsNotNone(value, f"配置项缺失: sshout.{path}")

    def test_mention_patterns_validity(self):
        """测试提及模式的有效性"""
        patterns = self.config_manager.get('sshout.mention_patterns', [])

        self.assertIsInstance(patterns, list)
        self.assertGreater(len(patterns), 0, "提及模式列表不能为空")

        for pattern in patterns:
            with self.subTest(pattern=pattern):
                self.assertIsInstance(pattern, str)
                self.assertGreater(len(pattern), 0)
                # 验证模式包含 Claude
                self.assertIn('claude', pattern.lower())


if __name__ == '__main__':
    # 运行测试
    print("🧪 开始SSHOUT集成测试...")
    print("💡 提示：设置环境变量 TEST_REAL_SSHOUT=1 来运行实际连接测试")
    unittest.main(verbosity=2)