89631f9ddf
new file: package.py new file: packup.bat new file: packup.sh new file: requirements.txt new file: scripts/__init__.py new file: scripts/file_store_api.py new file: src/__init__.py new file: src/config.toml new file: src/modules/__init__.py new file: src/modules/plugin_modules.py new file: src/modules/user_module.py new file: src/process.py new file: test.py
142 lines
4.7 KiB
Python
142 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
import sys
|
||
import os
|
||
import importlib
|
||
from pathlib import Path
|
||
from typing import Type, List
|
||
from unittest.mock import Mock
|
||
|
||
# 添加src目录到系统路径
|
||
|
||
# 导入基础类
|
||
from src.modules.plugin_modules import MessageContext, BasePlugin
|
||
|
||
class PluginTester:
|
||
"""
|
||
插件测试助手类
|
||
创建模拟上下文和测试配置环境
|
||
"""
|
||
def __init__(self, plugin_class: Type[BasePlugin], config_dir: str = "test_configs"):
|
||
self.plugin_class = plugin_class
|
||
self.config_dir = config_dir
|
||
os.makedirs(self.config_dir, exist_ok=True)
|
||
|
||
# 创建模拟上下文
|
||
self.mock_ctx = MessageContext(
|
||
uid="test_user",
|
||
gid="test_group",
|
||
raw_message="测试消息",
|
||
id="test_bot"
|
||
)
|
||
|
||
# 创建插件实例
|
||
self.plugin = self.plugin_class(self.mock_ctx)
|
||
|
||
def test_before_load(self):
|
||
"""测试before_load方法（如果存在）"""
|
||
if hasattr(self.plugin, 'before_load'):
|
||
print(f"\n正在测试 {self.plugin_class.__name__} 的 before_load")
|
||
try:
|
||
result = self.plugin.before_load()
|
||
print(f"before_load 执行成功，返回值: {result}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"before_load 执行出错: {str(e)}")
|
||
return False
|
||
else:
|
||
print(f"\n{self.plugin_class.__name__} 中没有找到 before_load 方法")
|
||
return None
|
||
|
||
def test_after_load(self):
|
||
"""测试after_load方法（如果存在）"""
|
||
if hasattr(self.plugin, 'after_load'):
|
||
print(f"\n正在测试 {self.plugin_class.__name__} 的 after_load")
|
||
try:
|
||
result = self.plugin.after_load()
|
||
print(f"after_load 执行成功，返回值: {result}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"after_load 执行出错: {str(e)}")
|
||
return False
|
||
else:
|
||
print(f"\n{self.plugin_class.__name__} 中没有找到 after_load 方法")
|
||
return None
|
||
|
||
def test_after_save(self):
|
||
"""测试after_save方法（如果存在）"""
|
||
if hasattr(self.plugin, 'after_save'):
|
||
print(f"\n正在测试 {self.plugin_class.__name__} 的 after_save")
|
||
try:
|
||
result = self.plugin.after_save()
|
||
print(f"after_save 执行成功，返回值: {result}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"after_save 执行出错: {str(e)}")
|
||
return False
|
||
else:
|
||
print(f"\n{self.plugin_class.__name__} 中没有找到 after_save 方法")
|
||
return None
|
||
|
||
def find_plugin_classes(module_path: str) -> List[Type[BasePlugin]]:
|
||
"""
|
||
在process.py中查找所有继承自BasePlugin的类
|
||
返回类类型列表
|
||
"""
|
||
plugin_classes = []
|
||
|
||
# 动态导入模块
|
||
module_name = os.path.splitext(os.path.basename(module_path))[0]
|
||
spec = importlib.util.spec_from_file_location(module_name, module_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
# 查找所有继承自BasePlugin的类
|
||
for name, obj in vars(module).items():
|
||
try:
|
||
if isinstance(obj, type) and issubclass(obj, BasePlugin) and obj != BasePlugin:
|
||
plugin_classes.append(obj)
|
||
except TypeError:
|
||
continue
|
||
|
||
return plugin_classes
|
||
|
||
def main():
|
||
# process.py文件路径
|
||
process_path = Path("src/process.py")
|
||
|
||
if not process_path.exists():
|
||
print(f"错误: 在 {process_path} 没有找到 process.py")
|
||
return
|
||
|
||
# 查找process.py中的所有插件类
|
||
plugin_classes = find_plugin_classes(str(process_path))
|
||
|
||
if not plugin_classes:
|
||
print("在 process.py 中没有找到插件类")
|
||
return
|
||
|
||
print(f"发现了 {len(plugin_classes)} 个需要测试的插件类:")
|
||
for i, plugin_class in enumerate(plugin_classes, 1):
|
||
print(f"{i}. {plugin_class.__name__}")
|
||
|
||
# 测试每个插件类
|
||
for plugin_class in plugin_classes:
|
||
print(f"\n{'='*50}")
|
||
print(f"正在测试插件: {plugin_class.__name__}")
|
||
|
||
tester = PluginTester(plugin_class)
|
||
|
||
# 按照自然顺序测试生命周期方法
|
||
tester.test_before_load()
|
||
tester.test_after_load()
|
||
|
||
# 如果需要测试配置保存
|
||
if hasattr(plugin_class, 'after_save'):
|
||
tester.test_after_save()
|
||
|
||
print(f"{'='*50}\n")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|