Source code for testbot.resource.resource

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


[docs]__author__ = "Nuanguang Gu(Sunny)"
[docs]__email__ = "nuanguang.gu@aliyun.com"
import os import importlib from abc import ABCMeta, abstractmethod from testbot.result.logger import logger_manager from testbot.resource.modules.module import ModuleBase from testbot.config import MODULE_LOGS_PATH # 存放用户注册的配置接口对象类型
[docs]_resource_device_mapping = dict()
[docs]_resource_port_mapping = dict()
[docs]class ResourceError(Exception): def __init__(self, msg): super().__init__(msg)
[docs]def register_resource(category, resource_type, comm_callback): """ 注册配置接口实例化的方法或者类。 """ if category == "device": _resource_device_mapping[resource_type] = comm_callback elif category == "port": _resource_port_mapping[resource_type] = comm_callback
[docs]class Resource(metaclass=ABCMeta): """ 代表所有测试资源设备的配置类,字段动态定义 """
[docs] name: str
[docs] type: str
[docs] description: str
# 接口模块类列表
[docs] MODULES = [ "testbot.resource.modules.device_module.PowerModule" ]
def __new__(cls, *args, **kwargs): device_obj = object.__new__(cls) for mod in tuple(cls.MODULES): mod_pkg = ".".join(mod.split(".")[0:-1]) mod_clazz = mod.split(".")[-1] mod_cls = getattr(importlib.import_module(mod_pkg), mod_clazz) if not issubclass(mod_cls, ModuleBase): raise Exception(f"{mod_cls.__name__}类不是接口模块类ModuleBase的子类!请检查资源类{cls.__name__}的MODULES模块类列表!!") setattr(device_obj, mod_cls.__name__, mod_cls(resource=cls)) return device_obj @classmethod
[docs] def register_module(cls, module: str): """ 注册接口模块类 :param module: 接口模块类包路径 :type module: str :return: :rtype: """ if module not in cls.MODULES: cls.MODULES.append(module)
def __init__(self, name: str = "", *args, **kwargs): # logname = "".join(x for x in name if x.isalnum()) self.logger = kwargs.get("logger", logger_manager.register(logger_name="Resource", filename=os.path.join(MODULE_LOGS_PATH, "Resource.log"), for_test=True)) self.name = name self.type = kwargs.get("type", None) self.description = kwargs.get("description", None) self.pre_connect = False self.client_attributes = dict() self.shared_attributes = dict() self.server_attributes = dict() self.reserved = False @abstractmethod
[docs] def to_dict(self) -> dict: return dict()
@classmethod
[docs] def from_dict(cls): return None