From 786ac48d11778838018e32e8f346c9b0de3c92db Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Wed, 11 Feb 2026 13:41:50 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E9=80=9A=E8=BF=87=20pre-commit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/debug_verify.py | 38 ----- tests/unit/core/test_bot_manager_lazy_load.py | 91 ----------- tests/unit/core/test_bot_manager_metadata.py | 59 ------- tests/unit_test_platform.py | 105 ------------ tests/verify_docker.py | 133 --------------- tests/verify_history.py | 60 ------- tests/verify_phase3.py | 154 ------------------ 7 files changed, 640 deletions(-) delete mode 100644 tests/debug_verify.py delete mode 100644 tests/unit/core/test_bot_manager_lazy_load.py delete mode 100644 tests/unit/core/test_bot_manager_metadata.py delete mode 100644 tests/unit_test_platform.py delete mode 100644 tests/verify_docker.py delete mode 100644 tests/verify_history.py delete mode 100644 tests/verify_phase3.py diff --git a/tests/debug_verify.py b/tests/debug_verify.py deleted file mode 100644 index c594fa4..0000000 --- a/tests/debug_verify.py +++ /dev/null @@ -1,38 +0,0 @@ -import sys -import types -from unittest.mock import MagicMock - - -# Helper to mock packages -def mock_package(name): - m = types.ModuleType(name) - sys.modules[name] = m - return m - - -# Mock key dependencies -mock_package("mcp") -mock_package("mcp.server") -mock_package("mcp.server.fastmcp") -mock_package("google") -mock_package("google.genai") -mock_package("google.generativeai") -mock_package("openai") -mock_package("anthropic") -mock_package("httpx") - -# Adjust path to include the plugin directory -sys.path.append( - r"c:\Helianthus\astrpro\AstrBot-master\data\plugins\astrbot_plugin_qq_group_daily_analysis" -) -sys.path.append(r"c:\Helianthus\astrpro\AstrBot-master") - -print("Starting import...") -try: - from main import QQGroupDailyAnalysis - - print("Import successful") -except Exception: - import traceback - - traceback.print_exc() diff --git a/tests/unit/core/test_bot_manager_lazy_load.py b/tests/unit/core/test_bot_manager_lazy_load.py deleted file mode 100644 index d25bb97..0000000 --- a/tests/unit/core/test_bot_manager_lazy_load.py +++ /dev/null @@ -1,91 +0,0 @@ -import unittest -import sys -import os -from unittest.mock import MagicMock - -# Add paths -plugin_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) -if plugin_root not in sys.path: - sys.path.insert(0, plugin_root) -astrbot_root = os.path.abspath(os.path.join(plugin_root, "../../../")) -if astrbot_root not in sys.path: - sys.path.insert(0, astrbot_root) - -from src.core.bot_manager import BotManager # noqa: E402 -from src.infrastructure.platform.factory import PlatformAdapterFactory # noqa: E402 - -class TestBotManagerLazyLoad(unittest.TestCase): - def setUp(self): - self.config_manager = MagicMock() - self.config_manager.get_bot_qq_ids.return_value = [] - self.bot_manager = BotManager(self.config_manager) - - # Mock Context and PlatformManager - self.context = MagicMock() - self.platform_manager = MagicMock() - self.context.platform_manager = self.platform_manager - self.bot_manager.set_context(self.context) - - def test_lazy_load_discord(self): - """Test that BotManager lazily loads Discord adapter when client becomes ready later""" - - # 1. Setup a Mock Platform that is NOT ready yet (no client attribute or None) - mock_platform = MagicMock() - mock_platform.metadata.id = "discord" - mock_platform.metadata.name = "discord" - # Ensure it has NO client/bot attributes initially - del mock_platform.client - del mock_platform.bot - del mock_platform.get_client - - # Mock platform manager returning this platform - self.platform_manager.get_insts.return_value = [mock_platform] - - # 2. Run auto discovery (Simulate on_platform_loaded) - # Since it's async, we run it synchronously - import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - discovered = loop.run_until_complete(self.bot_manager.auto_discover_bot_instances()) - - # Verify it was "discovered" (added to dict) but likely as a placeholder or not fully init - # Based on my code: discovered[platform_id] = platform (if client missing) - self.assertIn("discord", discovered) - self.assertEqual(discovered["discord"], mock_platform) - - # Verify internal state: stored in _platforms but NOT in _bot_instances yet - self.assertIn("discord", self.bot_manager._platforms) - self.assertNotIn("discord", self.bot_manager._bot_instances) - - # 3. Simulate Client becoming ready - mock_client = MagicMock() # The DiscordBotClient - mock_platform.client = mock_client # Now it has the client - - # 4. Call get_bot_instance - should trigger lazy load - # We need to mock PlatformAdapterFactory to support "discord" and return a mock adapter - # Actually factory already supports it, but we need to ensure it doesn't fail on creation - # The create method takes (platform_name, bot_instance, config) - - # We assume "discord" is registered (it is in factory.py) - # But we need to make sure DiscordAdapter can be instantiated with our mock client - # DiscordAdapter needs discord module. If not present, it logs error. - # We should patch discord module if needed, but in Docker it exists. - - # Let's try calling get_bot_instance - instance = self.bot_manager.get_bot_instance("discord") - - # 5. Verify instance is returned and stored - self.assertIsNotNone(instance) - self.assertEqual(instance, mock_client) # get_bot_instance returns the bot_client (not adapter) - - # Verify it's now in _bot_instances - self.assertIn("discord", self.bot_manager._bot_instances) - - # Verify adapter was created - self.assertIn("discord", self.bot_manager._adapters) - - loop.close() - -if __name__ == "__main__": - unittest.main() diff --git a/tests/unit/core/test_bot_manager_metadata.py b/tests/unit/core/test_bot_manager_metadata.py deleted file mode 100644 index f62480a..0000000 --- a/tests/unit/core/test_bot_manager_metadata.py +++ /dev/null @@ -1,59 +0,0 @@ -import unittest -import sys -import os -from unittest.mock import MagicMock - -# Add paths -plugin_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) -if plugin_root not in sys.path: - sys.path.insert(0, plugin_root) -astrbot_root = os.path.abspath(os.path.join(plugin_root, "../../../")) -if astrbot_root not in sys.path: - sys.path.insert(0, astrbot_root) - -from src.core.bot_manager import BotManager # noqa: E402 - -class TestBotManagerMetadata(unittest.TestCase): - def setUp(self): - self.config_manager = MagicMock() - self.config_manager.get_bot_qq_ids.return_value = [] - self.bot_manager = BotManager(self.config_manager) - - self.context = MagicMock() - self.platform_manager = MagicMock() - self.context.platform_manager = self.platform_manager - self.bot_manager.set_context(self.context) - - def test_metadata_retrieval_via_meta_method(self): - """Test retrieving metadata via meta() method if attribute is missing""" - - # Mock Platform with NO metadata attribute, but has meta() method - mock_platform = MagicMock() - del mock_platform.metadata # Ensure no attribute - - mock_meta = MagicMock() - mock_meta.id = "discord_instance_1" - mock_meta.type = "discord" - mock_meta.name = "MyDiscordBot" - - mock_platform.meta.return_value = mock_meta - - # Setup get_insts - self.platform_manager.get_insts.return_value = [mock_platform] - - # Run auto_discover - import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - discovered = loop.run_until_complete(self.bot_manager.auto_discover_bot_instances()) - - # Verify it was discovered - self.assertIn("discord_instance_1", discovered) - # It should be in _platforms - self.assertIn("discord_instance_1", self.bot_manager._platforms) - - loop.close() - -if __name__ == "__main__": - unittest.main() diff --git a/tests/unit_test_platform.py b/tests/unit_test_platform.py deleted file mode 100644 index f5d1ce9..0000000 --- a/tests/unit_test_platform.py +++ /dev/null @@ -1,105 +0,0 @@ -import unittest -import sys -import os -from unittest.mock import MagicMock, AsyncMock - -# Add plugin root to path so we can import src -# Assuming this file is in tests/ -plugin_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) -if plugin_root not in sys.path: - sys.path.insert(0, plugin_root) - -# Add AstrBot root to path so we can import astrbot.api -astrbot_root = os.path.abspath(os.path.join(plugin_root, "../../../")) -if astrbot_root not in sys.path: - sys.path.insert(0, astrbot_root) - -print(f"Added to sys.path: {plugin_root}, {astrbot_root}") - -try: - from src.infrastructure.platform.factory import PlatformAdapterFactory # noqa: E402 - from src.infrastructure.platform.adapters.discord_adapter import DiscordAdapter # noqa: E402 -except ImportError as e: - print(f"Import Error: {e}") - print(f"sys.path: {sys.path}") - # Try importing as package - try: - from astrbot_plugin_qq_group_daily_analysis.src.infrastructure.platform.factory import PlatformAdapterFactory - # ... and others - except ImportError: - pass - raise e - -class TestPlatformArchitecture(unittest.TestCase): - - def setUp(self): - # Reset factory for isolation if needed, though hard with class methods - pass - - def test_discord_adapter_capabilities(self): - """测试 Discord 适配器能力配置""" - adapter = DiscordAdapter(bot_instance=MagicMock(), config={"bot_user_id": "123"}) - caps = adapter.get_capabilities() - - self.assertEqual(caps.platform_name, "discord") - self.assertTrue(caps.supports_message_history) - self.assertTrue(caps.supports_image_message) - # Check correct attribute name and value (30 from predefined caps) - self.assertEqual(caps.max_message_history_days, 30) - - def test_factory_registration(self): - """测试适配器工厂注册机制""" - # Discord should be registered by import - self.assertTrue(PlatformAdapterFactory.is_supported("discord")) - self.assertTrue(PlatformAdapterFactory.is_supported("aiocqhttp")) - - # Test creation - adapter = PlatformAdapterFactory.create("discord", MagicMock(), {}) - self.assertIsInstance(adapter, DiscordAdapter) - - def test_discord_fetch_messages(self): - """测试 Discord 消息获取逻辑 (Mocked)""" - # Mock bot instance - mock_bot = MagicMock() - mock_channel = MagicMock() - mock_bot.get_channel.return_value = mock_channel - - # Mock message history - # Create a mock message that mimics discord.Message - mock_msg = MagicMock() - mock_msg.id = 12345 - mock_msg.content = "test message" - mock_msg.author.id = 999 - mock_msg.author.name = "User" - mock_msg.created_at.timestamp.return_value = 1600000000 - mock_msg.attachments = [] - mock_msg.embeds = [] - mock_msg.stickers = [] - mock_msg.reference = None - - # history returns an async iterator - async def async_iter(): - yield mock_msg - - mock_channel.history.return_value = async_iter() - - # Initialize adapter - adapter = DiscordAdapter(bot_instance=mock_bot, config={"bot_user_id": "123"}) - - # Run async test - import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - messages = loop.run_until_complete( - adapter.fetch_messages("1001", days=1) - ) - - self.assertEqual(len(messages), 1) - self.assertEqual(messages[0].text_content, "test message") - self.assertEqual(messages[0].platform, "discord") - - loop.close() - -if __name__ == "__main__": - unittest.main() diff --git a/tests/verify_docker.py b/tests/verify_docker.py deleted file mode 100644 index 02d0723..0000000 --- a/tests/verify_docker.py +++ /dev/null @@ -1,133 +0,0 @@ -import sys -import asyncio -import unittest -from unittest.mock import MagicMock, AsyncMock, patch - -# Add paths for Docker environment -sys.path.append("/AstrBot") -sys.path.append("/AstrBot/data/plugins") # Ensure we can import the plugin as a package - -# Import real modules using package path -try: - from astrbot_plugin_qq_group_daily_analysis.src.scheduler.auto_scheduler import ( - AutoScheduler, - ) - from astrbot_plugin_qq_group_daily_analysis.main import QQGroupDailyAnalysis -except ImportError as e: - print(f"ImportError in Docker: {e}") - # Fallback to local paths if testing locally - sys.path.append(r"c:\Helianthus\astrpro\AstrBot-master\data\plugins") - from astrbot_plugin_qq_group_daily_analysis.src.scheduler.auto_scheduler import ( - AutoScheduler, - ) - from astrbot_plugin_qq_group_daily_analysis.main import QQGroupDailyAnalysis - - -class TestPhase3Docker(unittest.IsolatedAsyncioTestCase): - async def test_schedule_jobs(self): - """Test that schedule_jobs correctly adds jobs to APScheduler.""" - # Mocks for Logic only - config_manager = MagicMock() - config_manager.get_enable_auto_analysis.return_value = True - config_manager.get_auto_analysis_time.return_value = ["09:00", "18:30"] - - context = MagicMock() - scheduler = MagicMock() - context.cron_manager.scheduler = scheduler - - # Instantiate AutoScheduler - auto_scheduler = AutoScheduler( - config_manager, - MagicMock(), # message_handler - MagicMock(), # analyzer - MagicMock(), # report_generator - MagicMock(), # bot_manager - MagicMock(), # retry_manager - ) - - # Call schedule_jobs - auto_scheduler.schedule_jobs(context) - - # Verify add_job was called twice - self.assertEqual(scheduler.add_job.call_count, 2) - - # Verify call arguments - args, kwargs = scheduler.add_job.call_args_list[0] - self.assertEqual( - kwargs["id"], "astrbot_plugin_qq_group_daily_analysis_trigger_0" - ) - # Verify trigger via string representation to avoid attribute errors with real CronTrigger objects - trigger_str = str(kwargs["trigger"]) - self.assertTrue( - "hour='9'" in trigger_str or "hour=9" in trigger_str, - f"Trigger mismatch: {trigger_str}", - ) - self.assertTrue( - "minute='0'" in trigger_str or "minute=0" in trigger_str, - f"Trigger mismatch: {trigger_str}", - ) - - args, kwargs = scheduler.add_job.call_args_list[1] - self.assertEqual( - kwargs["id"], "astrbot_plugin_qq_group_daily_analysis_trigger_1" - ) - trigger_str = str(kwargs["trigger"]) - self.assertTrue( - "hour='18'" in trigger_str or "hour=18" in trigger_str, - f"Trigger mismatch: {trigger_str}", - ) - self.assertTrue( - "minute='30'" in trigger_str or "minute=30" in trigger_str, - f"Trigger mismatch: {trigger_str}", - ) - - async def test_main_on_platform_loaded(self): - """Test that main.py's on_platform_loaded triggers job scheduling.""" - context = MagicMock() - config = MagicMock() - - # Patch dependencies in main.py using full package path - with ( - patch( - "astrbot_plugin_qq_group_daily_analysis.main.ConfigManager" - ) as MockConfigManager, - patch( - "astrbot_plugin_qq_group_daily_analysis.main.BotManager" - ) as MockBotManager, - patch( - "astrbot_plugin_qq_group_daily_analysis.main.MessageAnalyzer" - ), - patch( - "astrbot_plugin_qq_group_daily_analysis.main.ReportGenerator" - ), - patch( - "astrbot_plugin_qq_group_daily_analysis.main.RetryManager" - ), - patch( - "astrbot_plugin_qq_group_daily_analysis.main.AutoScheduler" - ) as MockAutoScheduler, - ): - mock_auto_scheduler_instance = MockAutoScheduler.return_value - mock_config_instance = MockConfigManager.return_value - mock_config_instance.get_enable_auto_analysis.return_value = True - - mock_bot_manager_instance = MockBotManager.return_value - mock_bot_manager_instance.initialize_from_config = AsyncMock( - return_value={"test_platform": MagicMock()} - ) - - # Instantiate plugin - plugin = QQGroupDailyAnalysis(context, config) - - # Call on_platform_loaded - event = MagicMock() - await plugin.on_platform_loaded(event) - - # Verify initialized_from_config called - mock_bot_manager_instance.initialize_from_config.assert_called_once() - # Verify schedule_jobs called - mock_auto_scheduler_instance.schedule_jobs.assert_called_once_with(context) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/verify_history.py b/tests/verify_history.py deleted file mode 100644 index 0ca3f14..0000000 --- a/tests/verify_history.py +++ /dev/null @@ -1,60 +0,0 @@ -import asyncio -import sys -from unittest.mock import AsyncMock, MagicMock -from datetime import datetime - -# 模拟环境 -sys.modules['astrbot'] = MagicMock() -sys.modules['astrbot.api'] = MagicMock() - -async def verify_history(): - print("开始验证 HistoryManager...") - - # 模拟 Star 实例 - mock_star = MagicMock() - mock_star.put_kv_data = AsyncMock() - mock_star.get_kv_data = AsyncMock() - - # 导入 HistoryManager (使用相对路径或动态导入) - from src.core.history_manager import HistoryManager - - hm = HistoryManager(mock_star) - - # 1. 测试保存分析 - group_id = "123456" - mock_stats = MagicMock() - mock_stats.message_count = 100 - mock_stats.participant_count = 10 - - analysis_result = { - "statistics": mock_stats, - "topics": [], - "user_titles": [] - } - - print(f"测试保存群 {group_id} 的分析摘要 (带时间槽)...") - await hm.save_analysis(group_id, analysis_result, "2026-02-08", "12-00") - - # 验证 put_kv_data 被调用，且 key 正确 - expected_key = f"analysis_{group_id}_2026-02-08_12-00" - mock_star.put_kv_data.assert_called_once() - actual_key = mock_star.put_kv_data.call_args[0][0] - actual_data = mock_star.put_kv_data.call_args[0][1] - - assert actual_key == expected_key - assert actual_data["message_count"] == 100 - print(f"✅ 保存验证成功: Key={actual_key}") - - # 2. 测试检查历史 (不同时间点) - mock_star.get_kv_data.side_effect = lambda type, pid, key, default: {"message_count": 100} if "12-00" in key else None - - has_history_12 = await hm.has_history(group_id, "2026-02-08", "12-00") - assert has_history_12 is True - print(f"✅ 12:00 历史检查成功") - - has_history_13 = await hm.has_history(group_id, "2026-02-08", "13-00") - assert has_history_13 is False - print(f"✅ 13:00 (无历史) 检查成功") - -if __name__ == "__main__": - asyncio.run(verify_history()) diff --git a/tests/verify_phase3.py b/tests/verify_phase3.py deleted file mode 100644 index e4b8571..0000000 --- a/tests/verify_phase3.py +++ /dev/null @@ -1,154 +0,0 @@ -import sys -import types -from unittest.mock import MagicMock - - -# Helper to mock packages -def mock_package(name): - m = types.ModuleType(name) - m.__path__ = [] - m.__spec__ = None # Fix for importlib - m.__loader__ = None - sys.modules[name] = m - return m - - -# Mock key dependencies with proper linking -mcp = mock_package("mcp") -mcp_server = mock_package("mcp.server") -mcp_server_fastmcp = mock_package("mcp.server.fastmcp") -mcp_types = mock_package("mcp.types") -# Link submodules -mcp.server = mcp_server -mcp.types = mcp_types -mcp_server.fastmcp = mcp_server_fastmcp - -# Add CallToolResult to mcp.types -mcp_types.CallToolResult = MagicMock() - -google = mock_package("google") -google_genai = mock_package("google.genai") -google_generativeai = mock_package("google.generativeai") -google.genai = google_genai -google.generativeai = google_generativeai - -openai = mock_package("openai") -openai_types = mock_package("openai.types") -openai.types = openai_types - -anthropic = mock_package("anthropic") -anthropic_types = mock_package("anthropic.types") -anthropic.types = anthropic_types - -httpx = mock_package("httpx") - -import asyncio # noqa: E402 -import unittest # noqa: E402 -from unittest.mock import AsyncMock, patch # noqa: E402 - -# Adjust path to include the plugin directory -sys.path.append( - r"c:\Helianthus\astrpro\AstrBot-master\data\plugins\astrbot_plugin_qq_group_daily_analysis" -) -sys.path.append(r"c:\Helianthus\astrpro\AstrBot-master") -# Add Docker container path -sys.path.append("/AstrBot") -sys.path.append("/AstrBot/data/plugins/astrbot_plugin_qq_group_daily_analysis") - -from src.scheduler.auto_scheduler import AutoScheduler # noqa: E402 -from main import QQGroupDailyAnalysis # noqa: E402 -from astrbot.api.event import filter # noqa: E402, F401 - - -class TestPhase3(unittest.IsolatedAsyncioTestCase): - async def test_schedule_jobs(self): - """Test that schedule_jobs correctly adds jobs to APScheduler.""" - # Mocks - config_manager = MagicMock() - config_manager.get_enable_auto_analysis.return_value = True - config_manager.get_auto_analysis_time.return_value = ["09:00", "18:30"] - - context = MagicMock() - scheduler = MagicMock() - context.cron_manager.scheduler = scheduler - - # Instantiate AutoScheduler (mocking dependencies) - auto_scheduler = AutoScheduler( - config_manager, - MagicMock(), # message_handler - MagicMock(), # analyzer - MagicMock(), # report_generator - MagicMock(), # bot_manager - MagicMock(), # retry_manager - ) - - # Call schedule_jobs - auto_scheduler.schedule_jobs(context) - - # Verify unschedule_jobs was called (scheduler.get_job called) - # Verify add_job was called twice - self.assertEqual(scheduler.add_job.call_count, 2) - - # Verify call arguments for first job - args, kwargs = scheduler.add_job.call_args_list[0] - # Check trigger args - self.assertEqual( - kwargs["id"], "astrbot_plugin_qq_group_daily_analysis_trigger_0" - ) - self.assertEqual(kwargs["trigger"].hour, 9) - self.assertEqual(kwargs["trigger"].minute, 0) - - # Verify call arguments for second job - args, kwargs = scheduler.add_job.call_args_list[1] - self.assertEqual( - kwargs["id"], "astrbot_plugin_qq_group_daily_analysis_trigger_1" - ) - self.assertEqual(kwargs["trigger"].hour, 18) - self.assertEqual(kwargs["trigger"].minute, 30) - - async def test_main_on_platform_loaded(self): - """Test that main.py's on_platform_loaded triggers job scheduling.""" - # Mock Context and Config - context = MagicMock() - config = MagicMock() - - # Mock dependencies inside QQGroupDailyAnalysis - # We need to patch the classes instantiated in __init__ - with ( - patch("main.ConfigManager") as MockConfigManager, - patch("main.BotManager") as MockBotManager, - patch("main.MessageAnalyzer"), - patch("main.ReportGenerator"), - patch("main.RetryManager") as MockRetryManager, - patch("main.AutoScheduler") as MockAutoScheduler, - patch("main.TraceLogFilter"), - ): - # Setup mocks - mock_auto_scheduler_instance = MockAutoScheduler.return_value - mock_config_instance = MockConfigManager.return_value - mock_config_instance.get_enable_auto_analysis.return_value = True - - mock_bot_manager_instance = MockBotManager.return_value - mock_bot_manager_instance.initialize_from_config = AsyncMock( - return_value={"test_platform": MagicMock()} - ) - - # Instantiate plugin - plugin = QQGroupDailyAnalysis(context, config) - - # Call on_platform_loaded - event = MagicMock() - await plugin.on_platform_loaded(event) - - # Verify initialized_from_config called - mock_bot_manager_instance.initialize_from_config.assert_called_once() - - # Verify schedule_jobs called - mock_auto_scheduler_instance.schedule_jobs.assert_called_once_with(context) - - # Verify retry_manager start called - MockRetryManager.return_value.start.assert_called_once() - - -if __name__ == "__main__": - unittest.main()