| """ |
| Webhook功能单元测试 |
| """ |
| |
| import pytest |
| import asyncio |
| import aiohttp |
| from unittest.mock import Mock, AsyncMock, patch |
| from datetime import datetime |
| |
| from claude_agent.webhook.models import ( |
| BotMessage, UserInfo, ReplyInfo, WebhookConfig, |
| BotRegistration, MessageBroadcast |
| ) |
| from claude_agent.webhook.server import WebhookServer, BotRegistry, MessageDistributor |
| from claude_agent.webhook.client import WebhookClient |
| from claude_agent.webhook.integration import BotWebhookIntegration |
| |
| |
| class TestWebhookModels: |
| """测试Webhook数据模型""" |
| |
| def test_user_info_creation(self): |
| """测试UserInfo创建""" |
| user_info = UserInfo( |
| user_id=123, |
| username="testuser", |
| first_name="Test", |
| last_name="User", |
| is_bot=False |
| ) |
| assert user_info.user_id == 123 |
| assert user_info.username == "testuser" |
| assert user_info.is_bot is False |
| |
| def test_bot_message_creation(self): |
| """测试BotMessage创建""" |
| user_info = UserInfo(user_id=123, username="testuser") |
| |
| message = BotMessage( |
| bot_username="testbot", |
| group_id=-123456, |
| message_content="Hello world", |
| sender_info=user_info |
| ) |
| |
| assert message.bot_username == "testbot" |
| assert message.group_id == -123456 |
| assert message.message_content == "Hello world" |
| assert message.sender_info.user_id == 123 |
| |
| def test_webhook_config_defaults(self): |
| """测试WebhookConfig默认值""" |
| config = WebhookConfig(auth_token="test-token") |
| |
| assert config.server_host == "0.0.0.0" |
| assert config.server_port == 8080 |
| assert config.auth_token == "test-token" |
| assert config.max_retries == 3 |
| |
| |
| class TestBotRegistry: |
| """测试Bot注册管理器""" |
| |
| def setup_method(self): |
| """设置测试环境""" |
| self.registry = BotRegistry() |
| |
| def test_register_bot(self): |
| """测试Bot注册""" |
| registration = BotRegistration( |
| bot_username="testbot", |
| auth_token="test-token", |
| subscribed_groups=[-123, -456] |
| ) |
| |
| result = self.registry.register_bot(registration) |
| assert result is True |
| |
| # 验证订阅关系 |
| subscribers = self.registry.get_subscribers_for_group(-123) |
| assert "testbot" in subscribers |
| |
| def test_unregister_bot(self): |
| """测试Bot注销""" |
| registration = BotRegistration( |
| bot_username="testbot", |
| auth_token="test-token", |
| subscribed_groups=[-123] |
| ) |
| |
| self.registry.register_bot(registration) |
| result = self.registry.unregister_bot("testbot") |
| assert result is True |
| |
| # 验证订阅关系已移除 |
| subscribers = self.registry.get_subscribers_for_group(-123) |
| assert "testbot" not in subscribers |
| |
| def test_get_subscribers_with_exclusion(self): |
| """测试获取订阅者时排除指定Bot""" |
| # 注册多个Bot |
| for i, username in enumerate(["bot1", "bot2", "bot3"]): |
| registration = BotRegistration( |
| bot_username=username, |
| auth_token="test-token", |
| subscribed_groups=[-123] |
| ) |
| self.registry.register_bot(registration) |
| |
| # 测试排除功能 |
| subscribers = self.registry.get_subscribers_for_group(-123, exclude_bots=["bot1"]) |
| assert "bot1" not in subscribers |
| assert "bot2" in subscribers |
| assert "bot3" in subscribers |
| |
| |
| class TestMessageDistributor: |
| """测试消息分发器""" |
| |
| def setup_method(self): |
| """设置测试环境""" |
| self.registry = BotRegistry() |
| self.config = WebhookConfig(auth_token="test-token") |
| self.distributor = MessageDistributor(self.registry, self.config) |
| |
| @pytest.mark.asyncio |
| async def test_distributor_lifecycle(self): |
| """测试分发器生命周期""" |
| await self.distributor.start() |
| assert self.distributor.session is not None |
| |
| await self.distributor.stop() |
| assert self.distributor.session is None |
| |
| @pytest.mark.asyncio |
| async def test_distribute_message_no_subscribers(self): |
| """测试分发消息但无订阅者""" |
| await self.distributor.start() |
| |
| try: |
| user_info = UserInfo(user_id=123, username="testuser") |
| message = BotMessage( |
| bot_username="sender_bot", |
| group_id=-123, |
| message_content="test message", |
| sender_info=user_info |
| ) |
| |
| broadcast = MessageBroadcast(bot_message=message) |
| results = await self.distributor.distribute_message(broadcast) |
| |
| assert len(results) == 0 |
| finally: |
| await self.distributor.stop() |
| |
| |
| class TestWebhookClient: |
| """测试Webhook客户端""" |
| |
| def setup_method(self): |
| """设置测试环境""" |
| self.config = WebhookConfig( |
| webhook_url="http://localhost:8080", |
| auth_token="test-token" |
| ) |
| self.client = WebhookClient( |
| config=self.config, |
| bot_username="testbot", |
| subscribed_groups=[-123], |
| message_handler=None |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_client_creation(self): |
| """测试客户端创建""" |
| assert self.client.bot_username == "testbot" |
| assert self.client.subscribed_groups == [-123] |
| assert self.client.is_registered is False |
| |
| @pytest.mark.asyncio |
| async def test_check_server_status_no_server(self): |
| """测试检查服务器状态（无服务器）""" |
| # 模拟没有运行的服务器 |
| await self.client.start() |
| try: |
| status = await self.client.check_server_status() |
| assert status is False |
| finally: |
| await self.client.stop() |
| |
| |
| class TestBotWebhookIntegration: |
| """测试Bot Webhook集成""" |
| |
| def setup_method(self): |
| """设置测试环境""" |
| self.telegram_client = Mock() |
| self.config = WebhookConfig( |
| webhook_url="http://localhost:8080", |
| auth_token="test-token" |
| ) |
| self.integration = BotWebhookIntegration( |
| telegram_client=self.telegram_client, |
| webhook_config=self.config, |
| bot_username="testbot", |
| subscribed_groups=[-123] |
| ) |
| |
| def test_integration_creation(self): |
| """测试集成创建""" |
| assert self.integration.bot_username == "testbot" |
| assert self.integration.subscribed_groups == [-123] |
| assert self.integration.is_running is False |
| |
| @pytest.mark.asyncio |
| async def test_process_bot_message_to_context(self): |
| """测试处理Bot消息到上下文""" |
| user_info = UserInfo(user_id=123, username="otherbot", is_bot=True) |
| message = BotMessage( |
| bot_username="otherbot", |
| group_id=-123, |
| message_content="Hello from other bot", |
| sender_info=user_info |
| ) |
| |
| # 测试消息处理不会抛出异常 |
| await self.integration._process_bot_message_to_context(message) |
| |
| def test_get_status(self): |
| """测试获取集成状态""" |
| status = self.integration.get_status() |
| |
| assert status["is_running"] is False |
| assert status["bot_username"] == "testbot" |
| assert status["subscribed_groups"] == [-123] |
| |
| |
| class TestWebhookIntegrationFlow: |
| """测试Webhook集成流程""" |
| |
| @pytest.mark.asyncio |
| async def test_message_flow_simulation(self): |
| """测试消息流程模拟""" |
| # 创建模拟的消息处理器 |
| received_messages = [] |
| |
| async def mock_message_handler(message: BotMessage): |
| received_messages.append(message) |
| |
| # 创建集成实例 |
| telegram_client = Mock() |
| config = WebhookConfig( |
| webhook_url="http://localhost:8080", |
| auth_token="test-token" |
| ) |
| integration = BotWebhookIntegration( |
| telegram_client=telegram_client, |
| webhook_config=config, |
| bot_username="testbot", |
| subscribed_groups=[-123] |
| ) |
| |
| # 模拟接收消息 |
| user_info = UserInfo(user_id=456, username="otherbot", is_bot=True) |
| message = BotMessage( |
| bot_username="otherbot", |
| group_id=-123, |
| message_content="Hello from other bot", |
| sender_info=user_info |
| ) |
| |
| # 测试消息处理 |
| await integration._handle_webhook_message(message) |
| |
| # 验证消息不会被处理（因为不在订阅列表） |
| # 这个测试主要验证代码不会抛出异常 |
| |
| |
| if __name__ == "__main__": |
| # 运行测试 |
| pytest.main([__file__, "-v"]) |