Source code for testbot.config.logicmodule

#!/usr/bin/env python
# -*- coding: utf-8 -*-


[docs]__author__ = "Nuanguang Gu(Sunny)"
[docs]__email__ = "nuanguang.gu@aliyun.com"
import os import json from enum import Enum from threading import Thread from importlib import import_module from abc import ABCMeta, abstractmethod from testbot.result.reporter import ResultReporter from testbot.resource.pool import ResourcePool from testbot.config import CONFIG_PATH from testbot.config.setting import static_setting, SettingBase
[docs]class ModuleType(Enum):
[docs] PRE = 1
[docs] PARALLEL = 2
[docs] POST =3
[docs]class ModuleBase(metaclass=ABCMeta): """ 逻辑配置模块的基类 """
[docs] module_type = None
[docs] priority = 99
def __init__(self, report: ResultReporter, resource: ResourcePool): self.reporter = report self.resource = resource self.thread = None @abstractmethod
[docs] def action(self): """ 实现该方法来实现模块的逻辑功能 """ pass
[docs] def do(self): if self.module_type == ModuleType.PARALLEL: self.thread = Thread(target=self.action) self.thread.start() else: self.action()
@abstractmethod
[docs] def stop(self): """ 实现该方法来实现模块逻辑功能的终止方法 """ pass
@static_setting.setting("LogicModule")
[docs]class ModuleSetting(SettingBase):
[docs] module_list_file = os.path.join(CONFIG_PATH, "modulelist.json")
[docs] module_setting_path = CONFIG_PATH
[docs]class ModuleManager: """ 配置模块的管理 """ def __init__(self): self.modules = dict()
[docs] def load(self): """ 从模块列表装载所有模块类 """ if not os.path.exists(ModuleSetting.module_list_file): # 如果没有找到模块配置文件,则不做任何操作 return with open(ModuleSetting.module_list_file) as file: obj = json.load(file) for item in obj['modules']: try: # 配置条目格式: # { # "name": "modulename", # "package":"module path", # "setting_file": "setting filename", # "setting_path": "setting path" # } module_name = item['name'] module_package = item['package'] setting_file = item.get("setting_file", None) setting_path = item.get('setting_path', ModuleSetting.module_setting_path) m = import_module(module_package) for element, value in m.__dict__.items(): if element == module_name: self.modules[module_name] = { "class": value, "setting_file": setting_file, "setting_path": setting_path } except Exception: pass
[docs] def add_module(self, module_class, setting_file=None, setting_path=None): """ 添加模块 """ obj = { "class": module_class, "setting_file": setting_file, "setting_path": setting_path } self.modules[module_class.__name__] = obj
[docs] def get_module_instances(self, module_type, result_reporter, resources): """ 获取模块的实例化列表 """ rv = list() for mkey, mvalue in self.modules.items(): print(mvalue['class'].module_type) print(module_type) if mvalue['class'].module_type.value == module_type.value: rv.append(mvalue['class'](result_reporter, resources)) return rv
[docs] def save(self): """ 保存所有模块到模块配置列表 """ obj = dict() obj['modules'] = list() for mkey, mvalue in self.modules.items(): obj['modules'].append({ "name": mkey, "package": mvalue["class"].__module__, "setting_file": mvalue['setting_file'], "setting_path": mvalue['setting_path'] }) file_dir = os.path.dirname(ModuleSetting.module_list_file) if not os.path.exists(file_dir): os.makedirs(file_dir) with open(ModuleSetting.module_list_file, "w") as file: json.dump(obj, file, indent=4)
[docs] def run_module(self, type): pass
[docs] def stop_module(self): pass
if __name__ == "__main__": from product.logicmodules.demo import DemoModule
[docs] mm = ModuleManager()
mm.load() mm.add_module(DemoModule) mm.save() mm.load() post_module = mm.get_module_instances(ModuleType.POST, None, None) print(post_module)