Spaces:
Runtime error
Runtime error
| """ | |
| 测试:Loop 插件 | |
| =============== | |
| 测试所有内置插件的注册和执行 | |
| """ | |
| import sys | |
| sys.path.insert(0, '.') | |
| import asyncio | |
| from plugins.base import LoopPlugin, PluginRegistry, LoopStage, LoopContext, PluginResult | |
| class DummyPlugin(LoopPlugin): | |
| """用于测试的虚拟插件""" | |
| def stage(self) -> LoopStage: | |
| return LoopStage.SUMMARIZE | |
| def name(self) -> str: | |
| return "dummy" | |
| async def run(self, ctx: LoopContext) -> PluginResult: | |
| return PluginResult(success=True, data={"dummy": True}, message="dummy ok") | |
| class TestPluginBase: | |
| """测试插件基类""" | |
| def test_plugin_registry_init(self): | |
| """注册中心初始化应为空""" | |
| registry = PluginRegistry() | |
| assert registry.get_all() == [] | |
| def test_register_plugin(self): | |
| """注册插件""" | |
| registry = PluginRegistry() | |
| plugin = DummyPlugin() | |
| registry.register(plugin) | |
| assert len(registry.get_all()) == 1 | |
| def test_get_plugins_by_stage(self): | |
| """按阶段获取插件""" | |
| registry = PluginRegistry() | |
| registry.register(DummyPlugin()) | |
| plugins = registry.get_plugins(LoopStage.SUMMARIZE) | |
| assert len(plugins) == 1 | |
| def test_unregister_plugin(self): | |
| """注销插件""" | |
| registry = PluginRegistry() | |
| registry.register(DummyPlugin()) | |
| registry.unregister("dummy") | |
| assert len(registry.get_all()) == 0 | |
| def test_plugin_result_defaults(self): | |
| """PluginResult 默认值""" | |
| result = PluginResult() | |
| assert result.success is True | |
| assert result.data == {} | |
| assert result.message == "" | |
| def test_plugin_result_custom(self): | |
| """PluginResult 自定义值""" | |
| result = PluginResult(success=False, data={"key": "val"}, message="error") | |
| assert result.success is False | |
| assert result.data["key"] == "val" | |
| assert result.message == "error" | |
| def test_loop_context_defaults(self): | |
| """LoopContext 默认值""" | |
| ctx = LoopContext() | |
| assert ctx.briefing == {} | |
| assert ctx.diet_records == [] | |
| assert ctx.errors == [] | |
| class TestFamilyQuizPlugins: | |
| """测试家庭问卷插件""" | |
| def test_family_recipe_quiz_plugin(self): | |
| """家庭菜谱检查插件""" | |
| from plugins.family_quiz import FamilyRecipeQuizPlugin | |
| plugin = FamilyRecipeQuizPlugin() | |
| assert plugin.stage() == LoopStage.FAMILY_QUIZ | |
| assert plugin.name() == "family_recipe_quiz" | |
| def test_weight_quiz_plugin(self): | |
| """体重询问检查插件""" | |
| from plugins.family_quiz import WeightQuizPlugin | |
| plugin = WeightQuizPlugin() | |
| assert plugin.stage() == LoopStage.FAMILY_QUIZ | |
| assert plugin.name() == "weight_quiz" | |
| class TestSummarizePlugins: | |
| """测试总结阶段插件""" | |
| def test_diet_summary_plugin(self): | |
| """饮食总结插件""" | |
| from plugins.diet_summary import DietSummaryPlugin | |
| plugin = DietSummaryPlugin() | |
| assert plugin.stage() == LoopStage.SUMMARIZE | |
| assert plugin.name() == "diet_summary" | |
| def test_weight_check_plugin(self): | |
| """体重检查插件""" | |
| from plugins.weight_check import WeightCheckPlugin | |
| plugin = WeightCheckPlugin() | |
| assert plugin.stage() == LoopStage.SUMMARIZE | |
| assert plugin.name() == "weight_check" | |
| def test_family_memory_plugin(self): | |
| """家庭记忆插件""" | |
| from plugins.family_memory import FamilyMemoryPlugin | |
| plugin = FamilyMemoryPlugin() | |
| assert plugin.stage() == LoopStage.SUMMARIZE | |
| assert plugin.name() == "family_memory" | |
| class TestAnalyzePlugins: | |
| """测试分析阶段插件""" | |
| def test_dri_analysis_plugin(self): | |
| """DRIs 分析插件""" | |
| from plugins.dri_analysis import DRIAnalysisPlugin | |
| plugin = DRIAnalysisPlugin() | |
| assert plugin.stage() == LoopStage.ANALYZE | |
| assert plugin.name() == "dri_analysis" | |
| class TestBriefPlugins: | |
| """测试简报阶段插件""" | |
| def test_briefing_generator_plugin(self): | |
| """简报生成插件""" | |
| from plugins.briefing_generator import BriefingGeneratorPlugin | |
| plugin = BriefingGeneratorPlugin() | |
| assert plugin.stage() == LoopStage.BRIEF | |
| assert plugin.name() == "briefing_generator" | |
| class TestThreeDayPlugins: | |
| """测试三天总结插件""" | |
| def test_three_day_summary_plugin(self): | |
| """三天总结插件""" | |
| from plugins.three_day_summary import ThreeDaySummaryPlugin | |
| plugin = ThreeDaySummaryPlugin() | |
| assert plugin.stage() == LoopStage.THREE_DAY | |
| assert plugin.name() == "three_day_summary" | |
| class TestConsolidatePlugins: | |
| """测试整理阶段插件""" | |
| def test_preset_writer_plugin(self): | |
| """预设写入插件""" | |
| from plugins.preset_writer import PresetWriterPlugin | |
| plugin = PresetWriterPlugin() | |
| assert plugin.stage() == LoopStage.CONSOLIDATE | |
| assert plugin.name() == "preset_writer" | |
| class TestPluginExecution: | |
| """测试插件执行""" | |
| def test_dummy_plugin_execution(self): | |
| """虚拟插件执行""" | |
| plugin = DummyPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_family_recipe_quiz_execution(self): | |
| """家庭菜谱检查执行""" | |
| from plugins.family_quiz import FamilyRecipeQuizPlugin | |
| plugin = FamilyRecipeQuizPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_weight_quiz_execution(self): | |
| """体重询问检查执行""" | |
| from plugins.family_quiz import WeightQuizPlugin | |
| plugin = WeightQuizPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_diet_summary_execution(self): | |
| """饮食总结执行""" | |
| from plugins.diet_summary import DietSummaryPlugin | |
| plugin = DietSummaryPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_weight_check_execution(self): | |
| """体重检查执行""" | |
| from plugins.weight_check import WeightCheckPlugin | |
| plugin = WeightCheckPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_family_memory_execution(self): | |
| """家庭记忆执行""" | |
| from plugins.family_memory import FamilyMemoryPlugin | |
| plugin = FamilyMemoryPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_dri_analysis_execution(self): | |
| """DRIs 分析执行""" | |
| from plugins.dri_analysis import DRIAnalysisPlugin | |
| plugin = DRIAnalysisPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_briefing_generator_execution(self): | |
| """简报生成执行""" | |
| from plugins.briefing_generator import BriefingGeneratorPlugin | |
| plugin = BriefingGeneratorPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_three_day_summary_execution(self): | |
| """三天总结执行""" | |
| from plugins.three_day_summary import ThreeDaySummaryPlugin | |
| plugin = ThreeDaySummaryPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |
| def test_preset_writer_execution(self): | |
| """预设写入执行""" | |
| from plugins.preset_writer import PresetWriterPlugin | |
| plugin = PresetWriterPlugin() | |
| ctx = LoopContext() | |
| result = asyncio.run(plugin.run(ctx)) | |
| assert result.success is True | |