自动化测试平台化[v1.0.0][事件的设计与实现]

淩亂°似流年 2022-11-24 14:12 149阅读 0赞

## 事件描述的基类 ##

> 事件的基类提供了基本的事件描述及事件逻辑，提供给测试开发人员继承并实现事件的具体逻辑，通常情况下事件描述包括几个方面，事件名称、事件的参数、事件的类型、相关的执行参数、是否需要锁、事件的结果，此外事件还应具备三个能力，事件预处理、事件执行、事件清理

*  事件名称：用于描述事件的具体行为，它需要测试开发者在实现自己的事件时被重载，应直接使用类变量
 *  事件的参数：通常是测试资源的实例，事件往往都是基于一定的测试对象的，所以需要通过事件参数将测试资源的实例传递给事件
 *  事件类型：一类事件是背景事件，用于模拟一些长时间固定或随机间隔的一些事件；另一类是非背景事件，一般有固定的执行次数，往往用于某个具体的测试点的验证
 *  相关的执行参数：实际执行事件时的参数，例如时间间隔、执行次数等
 *  是否需要锁：对于同一个事件而言，是否需要锁不一定是由这个事件本身决定的，而是执行者决定的，相同的事件既可以在无竞争的环境下执行，也可以在有竞争的环境下执行，因此这个参数必须是可配的
 *  事件的结果：既可以由开发者在实现事件的时候指定，也可以在执行事件流的过程中更改事件的状态

--------------------

*  事件预处理：提供给测试开发者实现事件预设条件的检查机制
 *  事件执行：提供给测试开发者重载并实现事件的具体逻辑
 *  事件的清理：事件执行后，不管事件执行是否正常，都需要执行清理操作

## 代码实例 ##

""" 事件驱动 事件的类定义 """
    
    from abc import ABCMeta, abstractmethod
    from core.result.logger import logger
    from threading import Lock
    from enum import Enum
    
    
    class EventStatus(Enum):
    	""" EventStatus枚举用于描述事件的执行状态 """
        IDlE = "Idle"
        WAITING = "Waiting"
        RUNNING = "Running"
        SUCCESS = "Success"
        FAIL = "Failure"
        ERROR = "Error"
    
    
    class EventBase(metaclass=ABCMeta):
        """ 事件的基类 """
        # 类变量，因为事件是通过实例化来工作的，所以通过类变量可以在各个实例之间使用一个锁对象
        event_lock = Lock() 
        name = ""
    
        def __init__(self, description="", **kwargs):
            self.description = description
            self.arguments = dict()
            self.job = None # 用来保存APScheduler生成的job对象
            self.interval = kwargs.get("interval", 0)
            self.result = EventStatus.IDlE
            self.back_ground = False
            self.loop_count = 0
            self.log = kwargs.get("log", logger.register(f"Event_{self.name}"))
            self.need_lock = False
            self.reporter = None
    
        def run(self):
        	""" run方法会被注册在APScheduler的job中，首先执行pre_check()方法，判断预处理方法是否完成 然后执行action抽象方法，测试开发者通过实现action方法来实现事件的逻辑 final方法用于事件执行的清理，测试开发者可以对其进行重载 """
            if self.need_lock:
                self.lock()
            try:
                if not self.pre_check():
                    self.log.error("Pre-check failed")
                    self.result = EventStatus.FAIL
                    return
                self.action()
            except Exception as ex:
                self.result = EventStatus.ERROR
                self.log.exception(ex)
            finally:
                try:
                    self.final()
                except Exception as ex:
                    self.log.exception(ex)
                if self.need_lock:
                    self.unlock()
    
        @abstractmethod
        def action(self):
            pass
    
        def pre_check(self):
            return True
    
        def final(self):
            pass
    
        def lock(self):
            self.__class__.event_lock.acquire()
    
        def unlock(self):
            self.__class__.event_lock.release()

event\_lock是类变量