| """ |
| Webhook模块单元测试 |
| """ |
| |
| import pytest |
| from unittest.mock import Mock, AsyncMock, patch, MagicMock |
| import asyncio |
| from datetime import datetime |
| |
| import sys |
| from pathlib import Path |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) |
| |
| from claude_agent.webhook.models import ( |
| BotMessage, BotRegistration, MessageBroadcast, WebhookConfig, WebhookResponse |
| ) |
| from claude_agent.webhook.client import WebhookClient |
| from claude_agent.webhook.server import WebhookServer |
| from claude_agent.webhook.telegram_integration import ( |
| TelegramWebhookIntegration, create_webhook_integration_from_config |
| ) |
| |
| |
| class TestWebhookModels: |
| """Webhook模型测试""" |
| |
| def test_bot_message_creation(self): |
| """测试BotMessage创建""" |
| message = BotMessage( |
| message_id="test_123", |
| group_id=12345, |
| content="Test message", |
| sender_info={ |
| "user_id": 999, |
| "username": "testbot", |
| "is_bot": True |
| }, |
| timestamp=datetime.now().isoformat(), |
| message_type="text" |
| ) |
| assert message.message_id == "test_123" |
| assert message.group_id == 12345 |
| assert message.content == "Test message" |
| assert message.sender_info["user_id"] == 999 |
| assert message.message_type == "text" |
| |
| def test_bot_registration_creation(self): |
| """测试BotRegistration创建""" |
| registration = BotRegistration( |
| bot_username="testbot", |
| auth_token="test_token", |
| subscribed_groups=[12345, 67890] |
| ) |
| assert registration.bot_username == "testbot" |
| assert registration.auth_token == "test_token" |
| assert 12345 in registration.subscribed_groups |
| assert 67890 in registration.subscribed_groups |
| |
| def test_webhook_config_creation(self): |
| """测试WebhookConfig创建""" |
| config = WebhookConfig( |
| webhook_url="http://localhost:8080", |
| auth_token="test_token", |
| enabled_groups=[12345], |
| connection_timeout=10.0, |
| request_timeout=5.0 |
| ) |
| assert config.webhook_url == "http://localhost:8080" |
| assert config.auth_token == "test_token" |
| assert config.enabled_groups == [12345] |
| assert config.connection_timeout == 10.0 |
| |
| def test_message_broadcast_creation(self): |
| """测试MessageBroadcast创建""" |
| message = BotMessage( |
| message_id="test_123", |
| group_id=12345, |
| content="Test message", |
| sender_info={"user_id": 999}, |
| timestamp=datetime.now().isoformat() |
| ) |
| broadcast = MessageBroadcast( |
| bot_message=message, |
| target_groups=[12345], |
| exclude_bots=["excludebot"] |
| ) |
| assert broadcast.bot_message == message |
| assert broadcast.target_groups == [12345] |
| assert broadcast.exclude_bots == ["excludebot"] |
| |
| |
| class TestWebhookClient: |
| """Webhook客户端测试""" |
| |
| @pytest.fixture |
| def webhook_config(self): |
| """创建测试配置""" |
| return WebhookConfig( |
| webhook_url="http://localhost:8080", |
| auth_token="test_token", |
| enabled_groups=[12345], |
| connection_timeout=10.0, |
| request_timeout=5.0 |
| ) |
| |
| @pytest.fixture |
| def webhook_client(self, webhook_config): |
| """创建测试客户端""" |
| return WebhookClient( |
| config=webhook_config, |
| bot_username="testbot", |
| subscribed_groups=[12345] |
| ) |
| |
| def test_client_initialization(self, webhook_client, webhook_config): |
| """测试客户端初始化""" |
| assert webhook_client.config == webhook_config |
| assert webhook_client.bot_username == "testbot" |
| assert webhook_client.subscribed_groups == [12345] |
| assert webhook_client.session is None |
| assert webhook_client.is_registered is False |
| |
| @pytest.mark.asyncio |
| async def test_start_client(self, webhook_client): |
| """测试客户端启动""" |
| with patch('aiohttp.ClientSession') as mock_session_class: |
| mock_session = Mock() |
| mock_session_class.return_value = mock_session |
| |
| with patch.object(webhook_client, '_register', new_callable=AsyncMock) as mock_register: |
| await webhook_client.start() |
| |
| assert webhook_client.session == mock_session |
| mock_register.assert_called_once() |
| |
| @pytest.mark.asyncio |
| async def test_stop_client(self, webhook_client): |
| """测试客户端停止""" |
| # 设置已注册状态 |
| webhook_client.is_registered = True |
| mock_session = Mock() |
| mock_session.close = AsyncMock() |
| webhook_client.session = mock_session |
| |
| with patch.object(webhook_client, '_unregister', new_callable=AsyncMock) as mock_unregister: |
| await webhook_client.stop() |
| |
| mock_unregister.assert_called_once() |
| mock_session.close.assert_called_once() |
| assert webhook_client.session is None |
| |
| @pytest.mark.asyncio |
| async def test_broadcast_message(self, webhook_client): |
| """测试消息广播""" |
| # 设置已注册状态 |
| webhook_client.is_registered = True |
| mock_session = Mock() |
| mock_response = Mock() |
| mock_response.status = 200 |
| mock_session.post.return_value.__aenter__.return_value = mock_response |
| webhook_client.session = mock_session |
| |
| message = BotMessage( |
| message_id="test_123", |
| group_id=12345, |
| content="Test message", |
| sender_info={"user_id": 999}, |
| timestamp=datetime.now().isoformat() |
| ) |
| |
| result = await webhook_client.broadcast_message(message) |
| |
| assert result is True |
| mock_session.post.assert_called_once() |
| |
| @pytest.mark.asyncio |
| async def test_broadcast_message_not_registered(self, webhook_client): |
| """测试未注册时广播消息""" |
| webhook_client.is_registered = False |
| |
| message = BotMessage( |
| message_id="test_123", |
| group_id=12345, |
| content="Test message", |
| sender_info={"user_id": 999}, |
| timestamp=datetime.now().isoformat() |
| ) |
| |
| result = await webhook_client.broadcast_message(message) |
| assert result is False |
| |
| @pytest.mark.asyncio |
| async def test_check_server_status(self, webhook_client): |
| """测试服务器状态检查""" |
| mock_session = Mock() |
| mock_response = Mock() |
| mock_response.status = 200 |
| mock_session.get.return_value.__aenter__.return_value = mock_response |
| webhook_client.session = mock_session |
| |
| result = await webhook_client.check_server_status() |
| assert result is True |
| |
| @pytest.mark.asyncio |
| async def test_check_server_status_failed(self, webhook_client): |
| """测试服务器状态检查失败""" |
| mock_session = Mock() |
| mock_response = Mock() |
| mock_response.status = 500 |
| mock_session.get.return_value.__aenter__.return_value = mock_response |
| webhook_client.session = mock_session |
| |
| result = await webhook_client.check_server_status() |
| assert result is False |
| |
| |
| class TestWebhookServer: |
| """Webhook服务器测试""" |
| |
| @pytest.fixture |
| def webhook_config(self): |
| """创建测试配置""" |
| return WebhookConfig( |
| webhook_url="http://localhost:8080", |
| auth_token="test_token", |
| enabled_groups=[12345], |
| connection_timeout=10.0, |
| request_timeout=5.0 |
| ) |
| |
| @pytest.fixture |
| def webhook_server(self, webhook_config): |
| """创建测试服务器""" |
| return WebhookServer(webhook_config) |
| |
| def test_server_initialization(self, webhook_server, webhook_config): |
| """测试服务器初始化""" |
| assert webhook_server.config == webhook_config |
| assert webhook_server.registered_bots == {} |
| assert webhook_server.app is not None |
| |
| def test_register_bot(self, webhook_server): |
| """测试Bot注册""" |
| registration = BotRegistration( |
| bot_username="testbot", |
| auth_token="test_token", |
| subscribed_groups=[12345] |
| ) |
| |
| webhook_server.register_bot(registration) |
| |
| assert "testbot" in webhook_server.registered_bots |
| assert webhook_server.registered_bots["testbot"] == registration |
| |
| def test_unregister_bot(self, webhook_server): |
| """测试Bot注销""" |
| registration = BotRegistration( |
| bot_username="testbot", |
| auth_token="test_token", |
| subscribed_groups=[12345] |
| ) |
| |
| webhook_server.register_bot(registration) |
| webhook_server.unregister_bot("testbot") |
| |
| assert "testbot" not in webhook_server.registered_bots |
| |
| def test_get_subscribed_bots(self, webhook_server): |
| """测试获取订阅指定群组的Bot""" |
| registration1 = BotRegistration( |
| bot_username="bot1", |
| auth_token="test_token", |
| subscribed_groups=[12345] |
| ) |
| registration2 = BotRegistration( |
| bot_username="bot2", |
| auth_token="test_token", |
| subscribed_groups=[67890] |
| ) |
| registration3 = BotRegistration( |
| bot_username="bot3", |
| auth_token="test_token", |
| subscribed_groups=[12345, 67890] |
| ) |
| |
| webhook_server.register_bot(registration1) |
| webhook_server.register_bot(registration2) |
| webhook_server.register_bot(registration3) |
| |
| subscribed = webhook_server.get_subscribed_bots(12345) |
| assert len(subscribed) == 2 |
| assert "bot1" in subscribed |
| assert "bot3" in subscribed |
| |
| @pytest.mark.asyncio |
| async def test_broadcast_to_bots(self, webhook_server): |
| """测试向Bot广播消息""" |
| registration = BotRegistration( |
| bot_username="testbot", |
| auth_token="test_token", |
| subscribed_groups=[12345], |
| webhook_endpoint="http://localhost:8081/webhook" |
| ) |
| webhook_server.register_bot(registration) |
| |
| message = BotMessage( |
| message_id="test_123", |
| group_id=12345, |
| content="Test message", |
| sender_info={"user_id": 999}, |
| timestamp=datetime.now().isoformat() |
| ) |
| |
| with patch('aiohttp.ClientSession') as mock_session_class: |
| mock_session = Mock() |
| mock_response = Mock() |
| mock_response.status = 200 |
| mock_session.post.return_value.__aenter__.return_value = mock_response |
| mock_session_class.return_value.__aenter__.return_value = mock_session |
| mock_session_class.return_value.__aexit__.return_value = AsyncMock() |
| |
| result = await webhook_server.broadcast_to_bots(message) |
| assert result is True |
| |
| |
| class TestTelegramWebhookIntegration: |
| """Telegram Webhook集成测试""" |
| |
| @pytest.fixture |
| def mock_telegram_client(self): |
| """创建模拟Telegram客户端""" |
| return Mock() |
| |
| @pytest.fixture |
| def webhook_config(self): |
| """创建测试配置""" |
| return WebhookConfig( |
| webhook_url="http://localhost:8080", |
| auth_token="test_token", |
| enabled_groups=[12345] |
| ) |
| |
| @pytest.fixture |
| def message_saver(self): |
| """创建消息保存回调""" |
| return AsyncMock() |
| |
| @pytest.fixture |
| def telegram_integration(self, mock_telegram_client, webhook_config, message_saver): |
| """创建Telegram集成""" |
| return TelegramWebhookIntegration( |
| telegram_client=mock_telegram_client, |
| webhook_config=webhook_config, |
| bot_username="testbot", |
| subscribed_groups=[12345], |
| message_saver=message_saver |
| ) |
| |
| def test_integration_initialization(self, telegram_integration, message_saver): |
| """测试集成初始化""" |
| assert telegram_integration.message_saver == message_saver |
| |
| def test_set_message_saver(self, telegram_integration): |
| """测试设置消息保存回调""" |
| new_saver = AsyncMock() |
| telegram_integration.set_message_saver(new_saver) |
| assert telegram_integration.message_saver == new_saver |
| |
| @pytest.mark.asyncio |
| async def test_save_webhook_message_to_context(self, telegram_integration, message_saver): |
| """测试保存Webhook消息到上下文""" |
| await telegram_integration._save_webhook_message_to_context( |
| chat_id="12345", |
| user_id=999, |
| message="Test message", |
| user_info={"username": "testuser"} |
| ) |
| |
| message_saver.assert_called_once_with("12345", 999, "Test message", False, {"username": "testuser"}) |
| |
| def test_create_integration_from_config_enabled(self): |
| """测试从配置创建集成（启用状态）""" |
| mock_client = Mock() |
| config_dict = { |
| 'enabled': True, |
| 'server_url': 'http://localhost:8080', |
| 'auth_token': 'test_token', |
| 'client': { |
| 'subscribed_groups': [12345] |
| } |
| } |
| |
| integration = create_webhook_integration_from_config( |
| telegram_client=mock_client, |
| webhook_config_dict=config_dict, |
| bot_username="testbot" |
| ) |
| |
| assert integration is not None |
| assert isinstance(integration, TelegramWebhookIntegration) |
| |
| def test_create_integration_from_config_disabled(self): |
| """测试从配置创建集成（禁用状态）""" |
| mock_client = Mock() |
| config_dict = { |
| 'enabled': False, |
| 'server_url': 'http://localhost:8080', |
| 'auth_token': 'test_token' |
| } |
| |
| integration = create_webhook_integration_from_config( |
| telegram_client=mock_client, |
| webhook_config_dict=config_dict, |
| bot_username="testbot" |
| ) |
| |
| assert integration is None |
| |
| def test_create_integration_from_config_missing_required(self): |
| """测试从配置创建集成（缺少必需配置）""" |
| mock_client = Mock() |
| config_dict = { |
| 'enabled': True, |
| 'server_url': 'http://localhost:8080' |
| # 缺少 auth_token |
| } |
| |
| integration = create_webhook_integration_from_config( |
| telegram_client=mock_client, |
| webhook_config_dict=config_dict, |
| bot_username="testbot" |
| ) |
| |
| assert integration is None |
| |
| def test_create_integration_from_config_no_subscribed_groups(self): |
| """测试从配置创建集成（无订阅群组）""" |
| mock_client = Mock() |
| config_dict = { |
| 'enabled': True, |
| 'server_url': 'http://localhost:8080', |
| 'auth_token': 'test_token', |
| 'client': { |
| 'subscribed_groups': [] |
| } |
| } |
| |
| integration = create_webhook_integration_from_config( |
| telegram_client=mock_client, |
| webhook_config_dict=config_dict, |
| bot_username="testbot" |
| ) |
| |
| assert integration is None |