可以借鉴学习的得物自动化平台执行器设计与实现笔记

随着公司接口自动化应用逐渐深入，老自动化方案的弊端日渐凸显（线下脚本&自动化框架 + Jenkins + 平台[调度 + 报表 + ...]），例如：技术栈&框架&三方库差异大、用户兼容性差、用例编写效率低、平台接入复杂、平台化适配性差、用例脚本不可控、用例维护成本高、执行耗时长等。为此，我们将自动化平台由“半平台化”转型为“全平台化”，实现了轻量高效、功能完备、使用简单、标准化程度高的自动化平台，支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例。在用例执行方面，新平台没有被传统的自动化框架所束缚，自研了更适合平台化的“自动化用例执行器”。自动化执行器是自动化平台自研的自动化用例执行器，负责具体执行平台编写的自动化用例和脚本，支持单独调试和按测试计划批量执行用例。主要提供串/并行跑用例、占位符、系统方法、环境变量(只读)、变量空间(读/写)、解释执行API.步骤、原生执行代码脚本等能力。执行器是参考了优秀接口测试工具（Jemerer、Postman、eolink、MeterSpher等）和主流单元测试框架（TestNG、PyTest、unittest等）后进行自主研发的“自动化用例-执行器”（Made in 得物）。

一、名词解释

核心词解释：

环境变量：用户常用配置(域名、库链接、Redis 链接、自定义配置、数据驱动配置等)，系统注入变量（运行环境、染色标、触发人、任务 ID 等）【用户态(只读)】。

变量空间：系统、用户存取变量的容器，主要用于用例的通信，分为“全局空间(g_vars)”和“局部空间(l_vars)”【用户态读/写】，变量空间会被绑定到对应的工作线程上方便实时读取。

系统变量：由系统主动注入到变量空间中的变量，以“_”开头【用户态(只读)】。

系统方法：由系统提供的工具类方法如：生成随机数、处理时间等常用方法。

断言方法：由系统提供的断言方法，断言时只能用此类方法，否则无法监听断言结果。

占位符：用于在页面上进行参数替换，支持从变量空间、环境变量中取值，或调用系统方法，支持语法表达式。

控制区域：用于区分用户可操作的区域，类似于操作系统的用户态和内核态。

执行器控制区域（内核态）：除了用户态，都是内核态。

用户控制区域（用户态）：用户可以写脚本的地方，用户可以填参数的地方。

用户态/内核态示意图。

脚本属于用户态。

请求体、接口 PATH 等也属于用户态。

只读数据保护机制
由于环境变量、系统变量、组件调用参数等信息都是基础的元数据，不能被用户态随意更改，否则会引发不可预知的问题，所以用户态操作元数据时必须加一层保护机制，保护机制如下：
用户态通过代理方式交互：

用户态通过代理操作执行器提供的数据。

环境变量代理（EnvVarsProxy）：只提供 get()方法，不提供 set() 和 remove()方法。

动态绑定拦截：解决动态语言特性问题。

        - 可读不可写
    """
    def __init__(self, env_vars_space: EnvVarsSpace):
        self.__env_vars_space = env_vars_space

    def __setattr__(self, name, value):
        # 拦截非法调用栈（保护关键数据）
        stacks = inspect.stack()
        source_stack: FrameInfo = stacks[1]
        if source_stack.filename != __file__:
            raise Exception("EnvVarsProxy 对象不可设置属性")

        super().__setattr__(name, value)

    def __getattribute__(self, name):
        if name == '_EnvVarsProxy__env_vars_space' or name == '__dict__':
            # 拦截非法调用栈（保护关键数据）
            stacks = inspect.stack()
            source_stack: FrameInfo = stacks[1]
            if source_stack.filename != __file__:
                raise Exception("EnvVarsProxy 对象不可访问受保护属性")

        return super().__getattribute__(name)

深层保护-引用类型数据保护：get 数据时使用深拷贝方式返回数据副本，即使用户态改了数据副本的内部信息，元数据中的信息不会被影响，则不会影响其他用户或组件的正常执行。

def get(self, key):
    value = self.__env_vars_dict.get(str(key))
    if value is not None:
        return copy.deepcopy(value)

    return value

变量空间代理（VarsProxy）：非系统变量不限制，拦截系统变量 set 和 remove 操作。

sys_funcs.get_call_param()：只提供本方法获取调用参数，返回深拷贝的副本数据。

二

平台化Case业务架构

自动化-平台侧

自动化-执行器

三

计划执行器

计划执行器是用于正式执行测试用例的执行器，负责前置加载任务所以需要的所有资源，包括用例、组件、环境变量等资源。然后按测试计划的用例编排方式，执行具体的用例，具体内容如下：
系统变量注入。

环境变量注入。

脚本前置写入：必须在主线程中前置写入，否则出现偶现无解的灵异现象。

脚本工作空间隔离设置：用于控制脚本中操作文件时的默认工作目录。

公共组件匹配：负责对公共组件进行引用匹配（用例中添加组件实际上是个拓展指针，指向组件实例）。

数据驱动：驱动变量注入【针对非常规项目】。

Dubbo 服务：开启 Dubbo 服务发现任务。

任务运行简报：开启任务运行简报上报任务，让用户可以看到运行时的用例运行状态。

执行计划前置脚本：触发脚本组件工作器（ScriptComponentWorker）执行“脚本组件”。

按批次并行执行用例组：批次间并行执行，批次内运行模式控制串/并行执行用例。

批次内拆分子批次-强制串行用例拆分到新的子批次。

常规用例-单条执行：触发用例工作器（CaseWorker）执行用例。

数据驱动用例-裂变为多条顺序执行：触发用例工作器（CaseWorker）执行用例。

执行计划后置脚本：触发脚本组件工作器（ScriptComponentWorker）执行“脚本组件”。

执行结果上报：由“报告收集器”根据“日志管理器”汇总每条用例的执行结果以及错误日志和过程日志，并收集接口调用记录，一同上报到自动化平台，根据情况自动触发分批或全量上报结方式。

四

用例&组件调试器

主要用于对用户或者组件进行实时调试，调试结果不落库，不会影响用例的成功率。调试器是精简版的计划执行器，只针对单个用例或者组件进行单条调试，相对更加轻量级。
五

工作器（Worker）

负责控制执行具体的用例和脚本组件，以及局部空(l_vars)间初始化与回收、系统变量注入等，主要包括“ScriptComponentWorker”和“CaseWorder”等工作器。
脚本组件工作器

脚本组件工作器（ScriptComponentWorker）负责控脚本组件的执行，动态绑定主步环境变量和归属项目环境变量，构造脚本组件(ScriptComponent)、柔性性初始化局部空(l_vars)、关键系统变量注入等。
由于脚本组件只会在主线程和用例线程中被执行，所以归属的执行线程空间中若已经存在局部空间，则直接复用。

def _init(self):
    """初初始"""
    self.component = ScriptComponent(self.task_md, self.raw_data, self)
    # 存储用例关键信息到局部变量的系统变量中
    # 若当前线程已绑定VarsGroup对象，则直接返回其中VarsSpace对象，否则进行初始化然后返回 
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    l_vars_space.set(VarsSpaceManager.SCRIPT_COMPO_ID_KEY, self.component.component_md.component_id)
    l_vars_space.set(VarsSpaceManager.SCRIPT_COMPO_NAME_KEY, self.component.component_md.name)
def _clear(self):
    """空间清理"""
    ScriptComponentUtilImp.set_call_param(None)  # 清空当前组件的参数调用
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    l_vars_space.remove(VarsSpaceManager.SCRIPT_COMPO_ID_KEY)
    l_vars_space.remove(VarsSpaceManager.SCRIPT_COMPO_NAME_KEY)

脚本组件（ScriptComponent）执行逻辑片段

通过反射动态导入执行脚本模块，然后进行组件调用检验&参数替换（替换入参中的占位符），动态绑定组件调用参数（用户态只读），然后调用脚本的 call 方法【def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):】。

def run_script(self):
    ...
    script_obj = importlib.import_module(script_model_path)  # 动态导入模块
    ...
    # 组件相关代理对象准备，组件内只能通过代理对象与执行器进行脚本，代理对象会进行安全拦截 
    env_vars_proxy: EnvVarsProxy = EnvVarsSpaceManager.get_env_vars_proxy(self.component_md.project_id)
    g_vars_proxy: VarsProxy = VarsSpaceManager.get_g_vars_proxy()
    l_vars_proxy: VarsProxy = VarsSpaceManager.get_l_vars_proxy()
    g_vars_space: VarsSpace = VarsSpaceManager.get_g_vars_space()
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    sys_funcs_proxy: SysFuncsProxy = SysFuncsManager.get_sys_funcs_proxy()
    asserts_proxy: AssertFuncsProxy = AssertManage.get_sys_funcs_proxy()

    # call_param参数的占位符替换 
    new_call_param = {}
    for _key, _value in self.component_md.call_param.items():
        if isinstance(_value, str):
            sub_code, _value = VarExtraction.param_substitute(
                self.component_worker.logger_group, g_vars_space, l_vars_space, env_vars_proxy, _value,
                fail_desc_prefix=f'{self.component_desc_info}替换调用参数')
            if not sub_code:
                return

        # 兼容老的数据
        elif isinstance(_value, (dict, list)):
            _ = json.dumps(_value, ensure_ascii=False)
            sub_code, _ = VarExtraction.param_substitute(
                self.component_worker.logger_group, g_vars_space, l_vars_space, env_vars_proxy, _,
                fail_desc_prefix=f'{self.component_desc_info}替换调用参数')
            if not sub_code:
                return

            # 还原类型-转换失败则放弃转换（减少组件报错率）
            try:
                _value = json.loads(_)
            except Exception as e:
                self.component_worker.logger_group.logger_proxy.warning(f'替换调用参数失败-替换成功后不符合JSON格式【{str(e)}】',
                                                                        _add_to_logger_space=False)
        new_call_param[_key] = _value

    # 对调用参数类型进行兜底转换-转换失败则放弃转换（减少组件报错率）
    for rule in self.component_md.param_rule:
        _key = rule['name']
        if _key in new_call_param and rule['type'] != 'string':
            _value = new_call_param[_key]
            if isinstance(_value, str):
                try:
                    new_call_param[_key] = json.loads(_value)
                except Exception as e:
                    self.component_worker.logger_group.logger_proxy.warning(f'替换调用参数失败-类型不匹配【{str(e)}】',
                                                                            _add_to_logger_space=False)

    # 动态绑定组件调用参数，组件中可通过 sys_funcs.get_call_param() 获取到调用参数 
    ScriptComponentUtilImp.set_call_param(new_call_param)
    try:
        script_obj.call(env_vars_proxy, g_vars_proxy, l_vars_proxy, sys_funcs_proxy, asserts_proxy,
                        self.component_worker.logger_group.logger_proxy)
    except AssertFail as e:
    ...

用例工作器

用例工作器（CaseWorder）负责控制用例执行，强制初始化局部空间(l_vars)、主项目环境变量绑定、构造测试用例(DataCase)、失败重试控制（用例维度）。

def run(self):
    for i in range(self._retry_count):
        self._set_case_status(False)
        if i > 0:
            # 等待 retry_sleep 秒后进行重试
            if self.data_case.data_case_md.retry_sleep > 0:
                time.sleep(self.data_case.data_case_md.retry_sleep)

            self.retry_i = i
            self.logger_group.logger_proxy.info(
                f'第 {i} 次重试用例，初始化日志组（已等待 {self.data_case.data_case_md.retry_sleep} 秒）',
                _add_to_logger_space=False)
            # 重置当前用例日志存储对象（实现失败重试日志隔离）
            self.logger_group = LoggerManager.init_logger_group(
                self.task_md, f'{self.task_md.runtime_type},case_id:{self.raw_data["son_step_id"]}')

        # 标记是否继续循环
        if_continue = False
        try:
            # 执行用例，失败则根据重试次数进行重试，成功则结束循环
            self._run()
            # 若存在执行错误日志则表示执行失败，重试执行
            if self.logger_group.logger_space.fail_msg_records:
                self._set_case_status(True)
                if_continue = True

        except AssertFail as e:
            # 这里获取到断言异常只可能是用例前置脚本的断言异常
            self.logger_group.logger_space.add_fail_msg(
                LogUnit(title=self.data_case.case_desc_info,
                        content=f'断言失败: {str(e)}',
                        newline=1))
            self.logger_group.logger_proxy.info(f'执行用例步骤-{str(e)}', _add_to_logger_space=False)
            self._set_case_status(True)
            self._run_after_script()
            continue
        ...
def _run(self):
    self._init()
    self.data_case.run_case()
def _init(self):
    self._fail_msg = ''
    self.data_case = DataCase(self.task_md, self.raw_data, self)
    # 每次运行都需要强制初始化，确保失败重试不被影响 
    VarsSpaceManager.init_l_vars_proxy()
    # 设置主步骤所在项目的项目ID
    EnvVarsSpaceManager.set_main_step_project_id(self.data_case.data_case_md.project_id)

测试用例（DataCase）执行逻辑片段

执行测试用例。

def run_case(self):
    """
    执行测试用例
        - 执行脚本组件，复用 ScriptComponentWorker 的能力     
        - case为失败状态时会执行剩下的有"is_finally"标的步骤    
            - "is_finally"标的步骤只有是断言失败时才影响case结果
    """
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    # 存储用例关键信息到局部变量的系统变量中
    l_vars_space.set(VarsSpaceManager.CASE_NAME_KEY, self.data_case_md.name)
    l_vars_space.set(VarsSpaceManager.CASE_ID_KEY, self.data_case_md.id)
    l_vars_space.set(VarsSpaceManager.CASE_AUTHOR_KEY, self.data_case_md.create_user)
    ...
    # 设置驱动数据到局部空间
    if self.case_worker.is_derive_case:
        drive_i_values: dict = self.case_worker.drive_info.get_drive_i_values(self.case_worker.drive_i)
        l_vars_proxy: VarsProxy = VarsSpaceManager.get_l_vars_proxy()
        for k, v in drive_i_values.items():
            # 替换驱动数据的值（替换占位符中的变量）
            sub_code, _value = self.param_substitute(v, f'{self._case_desc_info}替换驱动参数"{k}"的值')
            if not sub_code:
                return False
            l_vars_proxy.set(k, _value)

    res_code = self.run_before_script()
    if not res_code:
        return

    for case_step in self.data_case_md.case_steps:
        if case_step.step_type == DataCaseMd.API_H_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_http_step(case_step)
        elif case_step.step_type == DataCaseMd.API_D_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_dubbo_step(case_step)
        elif case_step.step_type == DataCaseMd.SCRIPT_COMPONENT_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            # 执行脚本组件
            self._run_script_component(case_step)
        elif case_step.step_type == DataCaseMd.API_G_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_grpc_step(case_step)
        elif case_step.step_type == DataCaseMd.SCRIPT_STEP_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_script_step(case_step)

执行 API.*步骤：

以 API.H(HttpStep)步骤为例，API.D(DubboStep)、API.G(GRpcStep) 步骤的主要逻辑基本相同。

def _run_http_step(self, case_step: HttpStepMd):
    from .steps.http_step import HttpStep
    retry_count = int(case_step.retry_count or 0) + 1 if self.task_md.allow_retry else 1
    retry_sleep = int(case_step.retry_sleep or 0)

    # 执行结果标
    run_res = False
    http_step = None
    for i in range(retry_count):
        if i > 0:
            if retry_sleep > 0:
                time.sleep(retry_sleep)

            # 转移错误日志到执行记录中 
            self.case_worker.logger_group.logger_space.move_fail_msg()

        http_step = HttpStep(self, case_step, retry_i=i)
        run_res = http_step.run_step()
        if run_res:
            return

    if not run_res:
        if not case_step.is_finally:
            self.is_fail = True
        elif http_step.assert_fail:
            # finally 步骤如果是断言失败，才会影响用例结果 
            self.is_fail = True
        else:
            # finally 步骤如果不是断言失败，则不影响用例结果，需要转移错误日志到执行记录中 
            self.case_worker.logger_group.logger_space.move_fail_msg()

查看全文：得物自动化平台执行器设计与实现