#!/usr/bin/env python
# -*- coding: utf-8 -*-
[docs]__copyright__ = "Copyright (c) 2024 Nuanguang Gu(Sunny) Reserved"
[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] 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