目录

一 名词解释

二 平台化Case业务架构

三 计划执行器

四 用例&组件调试器

五 工作器（Worker）

六 系统方法

七 占位符

八 内置库

九 总结

---

随着公司接口自动化应用逐渐深入，老自动化方案弊端日渐凸显（线下脚本&自动化框架 + Jenkins + 平台[调度 + 报表 + ...]），如：技术栈&框架&三方库差异大、用户兼容性差、用例编写效率低、平台接入复杂、平台化适配性差、用例脚本不可控、用例维护成本高、执行耗时长等。为此我们将自动化平台由“半平台化”转型为“全平台化”，实现了轻量高效、功能完备、使用简单、标准化程度高的自动化平台，支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例。在用例执行方面，新平台没有被传统的自动化框架所束缚，自研了更适合平台化的“自动化用例执行器”。

自动化执行器是自动化平台自研的自动化用例执行器，负责具体执行平台编写的自动化用例和脚本，支持单独调试和按测试计划批量执行用例。主要提供串/并行跑用例、占位符、系统方法、环境变量(只读)、变量空间(读/写)、解释执行API.步骤、原生执行代码脚本等能力。

执行器是参考了优秀接口测试工具（Jemerer、Postman、eolink、MeterSpher等）和主流单元测试框架（TestNG、PyTest、unittest等）后进行自主研发的“自动化用例-执行器”（Made in 得物）。

一 名词解释

核心词解释

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

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

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

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

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

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

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

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

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

• 用户态/内核态示意图。

  • 脚本属于用户态。

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

只读数据保护机制

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

用户态通过代理方式交互：

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

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

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

class EnvVarsProxy(object):    """    环境变量代理        - 可读不可写    """    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()

• http 步骤执行内容：

• • 准入条件校验。

  • 执行步骤的“前置脚本”。

  • 域名替换。

  • 参数替换：替换“api host、api path、params、headers、body、proxies、asserts”数据中的占位符变量。

# 替换headers参数new_headers = {}for _header in self.http_step_md.headers:    if _header.get('is_use') == '1' and _header.get('key_'):        _value = _header.get('value') or ''        sub_code, _value = self.data_case.param_substitute(str(_value), f'{self._step_desc_info}替换headers参数: ')        if not sub_code:            return False        new_headers[str(_header['key_'])] = _value

• • 接口调用：负责发送 http 请求。

def send_request(self, host: str, url: str, full_url: str, data: str = '', params=None, headers=None, timeout=None,                 proxies=None) -> Tuple[bool, Any]:    # 请求头追加字段    headers = self._handler_headers(headers)    ...    # 记录请求体数据    storage_body = data    try:        ...        if http_method in ['POST', 'PATCH', 'PUT', 'DELETE']:            # 处理请求体数据格式            handler_code, new_data, storage_body = self._handler_request_body(data, headers)            if not handler_code:                return False, None
            # 添加sign信息并更新url            if self._sign_info is not None:                self._api_request_response['full_url'] = full_url = self._url_add_sing(full_url, params)                self._logger_space.add_log(LogUnit(title='追加sign', content=full_url, indent=1),                                           log_type=LogInfo.DEBUG_LOG)
            self._api_request_response['request_body'] = storage_body
            if http_method == 'POST':                resp = self._send_post(host, url, data=new_data, params=params, headers=headers, timeout=timeout,                                       proxies=proxies)            elif http_method == 'PATCH':                resp = self._send_patch(host, url, data=new_data, params=params, headers=headers, timeout=timeout,                                        proxies=proxies)            elif http_method == 'PUT':                resp = self._send_put(host, url, data=new_data, params=params, headers=headers, timeout=timeout,                                      proxies=proxies)            else:                resp = self._send_delete(host, url, data=new_data, params=params, headers=headers, timeout=timeout,                                         proxies=proxies)
        elif http_method == 'GET':            resp = self._send_get(host, url, params=params, headers=headers, timeout=timeout, proxies=proxies)     ...

• 断言结果：按步骤中配置的断言项进行批量断言。

  • 期望值为数值类型时值 支持数学运算 如下案例:

    • 图中期望值为 int 类型，期望值表达式为【${_l->goods1->price} + 5 - ${_e->xxx->goods2->price}】。

      • 其中${_l->goods1->price} ：代表取出局部空间中的“goods1”下面的“price”的值，若为10。

      • 其中${_e->xxxx->goods2->price} ：代表取出环境变量中的“xxx”下面的“goods1”下面的“price”的值，若为3。

      • 则占位符变量替换后得到结果为【10 + 5 - 3】。

      • 数学运算后得到最终结果为【12】。

      • 最终将断言接口返回数据中的 $.data.price 是否 等于 12。

expect_value_type = str(assert_item.get('expect_value_type'))expect_value_func = self.EXPECT_VALUE_TYPE_DICT.get(expect_value_type)...# 期望值类型转换try:    expect_value = expect_value_func(assert_item.get('expect_value'))except Exception as e:...

def int_calculation(value: str):    """int类型转换，支持数学运行"""    if isinstance(value, int):        return value
    value = str(value)    if value.isnumeric():        return int(value)
    value = ast.literal_eval(value)    return int(value)

def float_calculation(value: str):    """float类型转换，支持数学运行"""    if isinstance(value, float):        return value
    value = str(value)    if value.isnumeric():        return float(value)
    value = ast.literal_eval(value)    return float(value)

• • 提取参数：将接口的返回数据按提取规则进行提取并存储到对应的变量空间中。

  • 执行步骤的“后置脚本”。

六 系统方法

由执行器提供的工具方法，供用户态通过代理（SysFuncsProxy）进行使用的工具类方法，可以在占位符和脚本代码中使用，如：生成随机数、处理时间等方法。

系统方法库

装载系统方法到方法库，装载后才可以在用户态的脚本和占位符中使用该方法。

@classmethoddef load_sys_func(cls, name='', param_rules: List[ParamRule] = None):    """    装载系统方法    :param name: 方法名    :param param_rules: 参数规则（注意：缺省值参数放到最后）    """    def decorator(func):        _name = name or func.__name__        if name in cls.SYS_FUNCS_DICT:            raise SysFuncsRepeatError(f'系统方法"{name}"重复')
        cls.SYS_FUNCS_DICT[_name] = {            "func": func,            "param_rules": param_rules or []        }        return func
    return decorator

系统方法实现

实现系统方法，需要通过@SysFuncsLib.load_sys_func()注入方法相关信息，以便调用规范校验和快捷模版生成。

• 案例：转换为时间戳(switch_timestamp)，该方法接收 3 个参数，且都是非缺省值参数。

@SysFuncsLib.load_sys_func(name='switch_timestamp', param_rules=[ParamRule('length', int, required=True),                                                                 ParamRule('date_string', str, required=True),                                                                 ParamRule('format', str, required=True)])def switch_timestamp(length: int, date_string: str, format: str) -> int:    """    转换为时间戳        - 将日期字符串转换为时间戳    :param length: 位数    :param date_string: 日期字符串    :param format: 日期格式    """    date_obj = datetime.datetime.strptime(date_string, format)    timestamp = str(date_obj.timestamp()).replace('.', '')
    # 补位    n = len(timestamp)    if n < length:        timestamp += '0' * (length - n)
    return int(timestamp[: length])

系统方法代理

class SysFuncsProxy(object):    """系统方法代理"""    ...    def __getattr__(self, name):        func_info = SysFuncsLib.SYS_FUNCS_DICT.get(name)        if not func_info:            raise SysFuncsNonExistentError(f'系统方法"{name}"不存在')
        return func_info['func']
    ...

系统方法在脚本中使用简介

• 执行器控制区域（内核态）：系统负责获取到（用户态）需要的所有代理对象，并触发调用脚本入口方法。

...# 相关代理对象准备 env_vars_proxy: EnvVarsProxy = EnvVarsSpaceManager.get_env_vars_proxy(self.data_case.data_case_md.project_id)g_vars_proxy: VarsProxy = VarsSpaceManager.get_g_vars_proxy()l_vars_proxy: VarsProxy = VarsSpaceManager.get_l_vars_proxy()sys_funcs_proxy: SysFuncsProxy = SysFuncsManager.get_sys_funcs_proxy()asserts_proxy: AssertFuncsProxy = AssertManage.get_sys_funcs_proxy()# 调用入口方法 script_obj.call(env_vars_proxy, g_vars_proxy, l_vars_proxy, sys_funcs_proxy,                asserts_proxy, self._logger_proxy)
...

• 用户控制区域（用户态）：在脚本代码中使用系统方法。

import jsonimport requests
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):    # 使用系统方法     res = sys_funcs.switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')    logger.info(res)

系统方法在占位符中使用介绍

• 用户控制区域（用户态）：在页面数据区域使用系统方法（如在请求体参数中使用）。

{    "timestamp": ${@switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')},    "xxx": "yyy"}

• 执行器控制区域（内核态）：系统负责提取原始数据中的方法调用占位符，解析调用语义，执行调用并替换结果。

  • 主要包括：

• • • 方法识别。

    • 调用方式校验：无参调用、有参调用。

    • 入参规则校验。

      • 从字符串中解析出对应的入参。

      • 参数个数、参数类型校验、必填参数校验。

      • 调用系统方法 & 结果替换。

• 系统方法占位符替换代码片段

...# 获取系统方法sys_func = sys_funcs_proxy.get_func(func_name)...# 获取系统参数规则param_rules: List[ParamRule] = sys_funcs_proxy.get_param_rules(func_name)# 无参调用 if not param_rules:    try:        value = sys_func()    except BaseException as e:        ...# 有参调用 else:    args_str: str = key[len(func_name) + 2:-1]  # 提取入参    if not args_str:        ...        return False, ''    raw_args = [str(_).strip() for _ in args_str.split(',')]    raw_args_n = len(raw_args)
    # 判断参数总个数正确性    if raw_args_n > len(param_rules):        ...        return False, ''
    required_n = 0    for param_rule in param_rules:        # param_rule.required 为 bool 类型，可以参与+法运行        required_n += param_rule.required
    # 判断必填参数个数正确    if raw_args_n < required_n:        ...        return False, ''
    # 生成系统方法入参    _args = []    for i, param_rule in enumerate(param_rules):        if param_rule.required and i + 1 > raw_args_n:            ...            return False, ''        elif i + 1 > raw_args_n:            # 未传部分缺省参数时结束循环            break
        try:            _raw_arg = raw_args[i]            if param_rule.param_type is bool:                # 布尔类型                if _raw_arg.capitalize() == 'True':                    _arg = True                elif _raw_arg.capitalize() == 'False':                    _arg = False            elif param_rule.param_type is str:                if len(_raw_arg) >= 2 and _raw_arg[0] == _raw_arg[-1] and _raw_arg[0] in ['"', "'"]:                    _arg = _raw_arg[1:-1]                else:                    ...                    return False, ''            else:                # 其他类型                _arg = param_rule.param_type(_raw_arg)        except Exception as e:            ...            return False, ''
        _args.append(_arg)
    try:        value = sys_func(*_args)    except BaseException as e:        ...        return False, ''

七 占位符

占位符是自动化平台使用频率很高的功能，提供页面数据区域与对应变量空间高效通信的能力。用于在页面上进行参数替换，支持从变量空间、环境变量中取值，或调用系统方法。分为普通占位符和增强占位符。

普通占位符

主要用于在页面数据区域获取变量空间中的变量值如 ${xxx}、${_case_id}，调用系统方法如 ${@get_trace_id()}。

增强占位符

支持使用语法表达式，可以准确、快捷、高效、多层级获取目标值。

• 拓展7项能力：

  • 支持直接取变量的属性，“->”代表属性类似于常规代码中的“.”

  • 支持直接取特殊类型(list，str，tuple)的索引和切片等复杂操作

  • 支持多层级取值

  • 支持取常规引用类型的属性，如：object.a

  • 占位符&脚本中支持快捷取值

  • 支持指定取对应空间的变量（局部、全局变量 或 环境变量）

• 拓展语法：

  • _l->：指定从局部变量取值

  • _g->：指定从全局变量取值

  • _e->：指定从环境变量取值

  • ->：取属性或key值（常规引用类型取属性，字段类型取key值）

  • []：索引或切片操作（支持list，str , tuple 等类型）（与Python本身的用法类似）

    • 示例：

      • [2]：取索引为2的元素

      • [-1]：取最后一位元素

      • [1:30]: 切片-取第2个到第30个

      • [1:30:2]：切片-取第2个到第30个，步长为2

      • [::-1]：切片（顺序颠倒）

【页面正使用占位符】

【脚本中使用增强key】

占位符使用案例

• 案例1："demoMobile": "${@random_fix_mobile("135")}"

  • 分析：存在一个普通占位符，调用系统方法random_fix_mobile("135"),该方法用于获取固定前缀随机手机号，支持用户手动传入 <= 13位的前缀，随机补充缺少的手机号位数，且整体符合手机号规范

  • 替换结果："demoMobile": "13579996688"

• 案例2："demoXx": ${orders[3]->labs[2:10:2]}

  • 分析：存在一个增强占位符，从局部空间或全局空间（优先局部空间）中获取订单列表(orders)中第4个订单的标签列表(labs)中第3到第10个元素步长为2

  • 替换结果："demoXx": ["lab3", "lab5", "lab7", "lab9"]

• 案例3："demoYy": "你好！${@random_zh(3)}，截止到${@switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')}时间，你在${_env}环境购买'${_e->goods.apple->name}'商品${_l->buy_times}次"。

• 分析：存在三个普通占位符和两个增强占位符。

  • ${@random_zh(3)}: 调用系统方法生成 3 个随机汉字。

  • ${@switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')}: 调用系统方法将日期格式转换成 13 位时间戳。

  • ${_env}: 获取系统变量中的当前执行环境。

  • ${_e->goods.apple->name}: 获取环境变量中“goods.apple”变量下的“name”字段的值。

  • ${_l->buy_times}：获取局部变量中“buy_times”变量的值。

  • 替换结果："demoYy": "你好！觙奲琍，截止到 1661134210000 时间，你在 t1 环境购买'苹果'商品 5 次"。

• 最终结果如下：

{  "batchNo": "1108000124800131",  "demoMobile": "13579996688",  "demoXx": ["lab3", "lab5", "lab7", "lab9"],  "demoYy": "你好！觙奲琍，截止到1661134210000时间，你在t1环境购买'苹果'商品5次",  ...}

核心代码片段

占位符 key 表达式解析

按占位符语法将 key 解析成对应的子 key 列表，如：${orders[3][2]->labs[2:10:2]} 将被解析成 5 个子表达式，即 sub_keys = ['orders', '[3]', '[2]', 'labs', '[2:10:2]']。

@classmethoddef parsing_sub_keys(cls, key: str) -> list:    """    占位符的key解析成子key列表    """    if not key:        return []        # 缓存加速    sub_keys = cls.KEY_MAP.get(key)    if sub_keys is not None:        return sub_keys
    raw_sub_keys = key.split('->')    sub_keys = []    # 解析表达式    for sub_key in raw_sub_keys:        if '[' in sub_key or ']' in sub_key:            if len(sub_key) <= 3 or sub_key.startswith('[') or not sub_key.endswith(']'):                raise ParseKeyException(f'表达式非法`{key}`')            if sub_key.count('[') != sub_key.count(']'):                raise ParseKeyException(f'表达式非法`{key}`')
            # 添加当前层的第一个字段名            _k0 = sub_key.split('[')[0]            sub_keys.append(_k0)
            _sub_key: str = sub_key[len(_k0):]            if '][' not in _sub_key:                if _sub_key.find('[') > _sub_key.find(']') or _sub_key.count('[') > 1 or _sub_key.count(']') > 1:                    raise ParseKeyException(f'表达式非法`{key}`')                sub_keys.append(_sub_key)            else:                _ = '<!-!>'  # 切割辅助符                while _ in _sub_key:                    _ += _  # 防止冲突
                _sub_keys = _sub_key.replace('][', f']{_}[').split(_)                for _sub_key in _sub_keys:                    if _sub_key.find('[') > _sub_key.find(']') or _sub_key.count('[') > 1 or _sub_key.count(']') > 1:                        raise ParseKeyException(f'表达式非法`{key}`')                    sub_keys.append(_sub_key)
        else:            sub_keys.append(sub_key)
    cls.KEY_MAP[key] = sub_keys    return sub_keys

占位符参数提取

根据提取到的子表达式（sub_keys）组层级匹配获取数据。

def get_var(self, key: str):    """获取变量值"""    sub_keys = self.parsing_sub_keys(key)    if not sub_keys:        return None
    sub_key_0 = sub_keys[0]    if sub_key_0 == '_g':        var, is_found = self.get_target_var(sub_keys[1:], self._g_vars, key)    elif sub_key_0 == '_l':        var, is_found = self.get_target_var(sub_keys[1:], self._l_vars, key)    elif sub_key_0 == '_e':        var, is_found = self.get_target_var(sub_keys[1:], self._env_vars, key)    else:        # 不指定变量空间时优先在 局部空间中查找，        var, is_found = self.get_target_var(sub_keys, self._l_vars, key)        if not is_found:            var, is_found = self.get_target_var(sub_keys, self._g_vars, key)
    return var
@classmethoddef get_target_var(cls, sub_keys: list, i_var: any, raw_key: str):    """    获取指定空间中的变量值    """    is_found = False    var = None    i = 0    for i, key in enumerate(sub_keys):        is_found = False        if i_var is None or not key:            break
        # 获取对象里的字段or属性        if not ('[' in key or ']' in key):            # 字典类型            if isinstance(i_var, dict):                is_found = key in i_var                i_var = i_var.get(key)
            # 获取空间中的变量值            elif isinstance(i_var, (VarsSpace, EnvVarsProxy)):                is_found = i_var.exist(key)                if is_found:                    i_var = i_var.get(key)                else:                    i_var = None
            # 获取其他类型对象的属性            else:                if hasattr(i_var, key):                    i_var = getattr(i_var, key)                    is_found = True                else:                    i_var = None
        # 获取列表、字符串、元组里的数据        else:            if type(i_var) not in [list, str, tuple, bytes, bytearray]:                raise ParseKeyException(f'表达式与实际变量值类型不匹配，无法通过索引取值`{raw_key}`')
            _index_temp = key[1:-1]            try:                _index_temp = int(_index_temp)            except Exception as e:                if 0 < _index_temp.count(':') <= 2:                    _i_temps = _index_temp.split(':')                    for _i, _i_temp in enumerate(_i_temps):                        try:                            if _i_temp != '':                                _i_temps[_i] = int(_i_temp)                            else:                                _i_temps[_i] = None                        except Exception as e:                            raise ParseKeyException(f'表达式非法`{raw_key}`，切片值只能是整数')                    if len(_i_temps) == 2:                        i_var = i_var[_i_temps[0]: _i_temps[1]]                        is_found = True                    else:                        i_var = i_var[_i_temps[0]: _i_temps[1]: _i_temps[2]]                        is_found = True                else:                    raise ParseKeyException(f'表达式非法`{raw_key}`')            else:                len_i_var = len(i_var)                if 0 <= len_i_var + _index_temp < 2 * len_i_var:                    i_var = i_var[_index_temp]                    is_found = True                else:                    raise ParseKeyException(f'表达式与实际变量值不匹配取值时索引越界(list index out of range)`{raw_key}`')
    if i == len(sub_keys) - 1 and is_found:        var = i_var
    return var, is_found

八 内置库

执行器中提供了丰富的内置库（Python 自带内置库、第三方库、平台自研内置库）来满足各种使用场景，可以在任何位置的脚本中导入使用。

九 总结

自动化执行器是自动化平台老方案中自动化框架的超集，提供框架能力的同时还负责解释执行自动化用例。技术实现上融入了优秀自动化框架的核心理念，也拓展支持了很多实用功能，同时更灵活、更具可塑性、更适配平台化，并支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例，也能满足低门槛、高效率产出自动化用例的需要，还能实现极其复杂的测试场景。在自动化平台转型后的近两年时间里，平台在质量保障、效率提升和使用体验方面效果显著，后续我们也将持续打磨完善执行器和自动化平台。