得物自动化平台执行器设计与实现
目录
一、名词解释
1. 核心词解释
2. 只读数据保护机制
二、平台化Case业务架构
三、计划执行器
四、用例&组件调试器
五、工作器（Worker）
1. 脚本组件工作器
2. 用例工作器
六、系统方法
1. 系统方法库
2. 系统方法实现
3. 系统方法代理
七、占位符
1. 普通占位符
2. 增强占位符
3. 占位符使用案例
4. 核心代码片段
八、内置库
九、总结
随着公司接口自动化应用逐渐深入，老自动化方案弊端日渐凸显（线下脚本&自动化框架 + Jenkins + 平台[调度 + 报表 + ...]），如：技术栈&框架&三方库差异大、用户兼容性差、用例编写效率低、平台接入复杂、平台化适配性差、用例脚本不可控、用例维护成本高、执行耗时长等。为此我们将自动化平台由“半平台化”转型为“全平台化”，实现了轻量高效、功能完备、使用简单、标准化程度高的自动化平台，支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例。在用例执行方面，新平台没有被传统的自动化框架所束缚，自研了更适合平台化的“自动化用例执行器”。
自动化执行器是自动化平台自研的自动化用例执行器，负责具体执行平台编写的自动化用例和脚本，支持单独调试和按测试计划批量执行用例。主要提供串/并行跑用例、占位符、系统方法、环境变量(只读)、变量空间(读/写)、解释执行API.步骤、原生执行代码脚本等能力。
执行器是参考了优秀接口测试工具（Jmeter、Postman、eolink、MeterSphere等）和主流单元测试框架（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_spacedef __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()：只提供本方法获取调用参数，返回深拷贝的副本数据。
自动化-平台侧
自动化-执行器
计划执行器是用于正式执行测试用例的执行器，负责前置加载任务需要的所有资源，包括用例、组件、环境变量等资源。然后按测试计划的用例编排方式，执行具体的用例，具体内容如下：
系统变量注入。
环境变量注入。
脚本前置写入：必须在主线程中前置写入，否则出现偶现无解的灵异现象。
脚本工作空间隔离设置：用于控制脚本中操作文件时的默认工作目录。
公共组件匹配：负责对公共组件进行引用匹配（用例中添加组件实际上是个拓展指针，指向组件实例）。
数据驱动：驱动变量注入【针对非常规项目】。
Dubbo 服务：开启 Dubbo 服务发现任务。
任务运行简报：开启任务运行简报上报任务，让用户可以看到运行时的用例运行状态。
执行计划前置脚本：触发脚本组件工作器（ScriptComponentWorker）执行“脚本组件”。
按批次并行执行用例组：批次间并行执行，批次内运行模式控制串/并行执行用例。
批次内拆分子批次-强制串行用例拆分到新的子批次。
常规用例-单条执行：触发用例工作器（CaseWorker）执行用例。
数据驱动用例-裂变为多条顺序执行：触发用例工作器（CaseWorker）执行用例。
执行计划后置脚本：触发脚本组件工作器（ScriptComponentWorker）执行“脚本组件”。
执行结果上报：由“报告收集器”根据“日志管理器”汇总每条用例的执行结果以及错误日志和过程日志，并收集接口调用记录，一同上报到自动化平台，根据情况自动触发分批或全量上报结方式。
主要用于对用例或者组件进行实时调试，调试结果不落库，不会影响用例的成功率。调试器是精简版的计划执行器，只针对单个用例或者组件进行单条调试，相对更加轻量级。
负责控制执行具体的用例和脚本组件，以及局部空间(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 = iself.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 = Falsetry:# 执行用例，失败则根据重试次数进行重试，成功则结束循环self._run()# 若存在执行错误日志则表示执行失败，重试执行if self.logger_group.logger_space.fail_msg_records:self._set_case_status(True)if_continue = Trueexcept 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()# 设置主步骤所在项目的项目IDEnvVarsSpaceManager.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 Falsel_vars_proxy.set(k, _value)res_code = self.run_before_script()if not res_code:returnfor 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:continueself._run_http_step(case_step)elif case_step.step_type == DataCaseMd.API_D_TYPE:if self.is_fail and not case_step.is_finally:continueself._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:continueself._run_grpc_step(case_step)elif case_step.step_type == DataCaseMd.SCRIPT_STEP_TYPE:if self.is_fail and not case_step.is_finally:continueself._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 HttpStepretry_count = int(case_step.retry_count or 0) + 1 if self.task_md.allow_retry else 1retry_sleep = int(case_step.retry_sleep or 0)# 执行结果标run_res = Falsehttp_step = Nonefor 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:returnif not run_res:if not case_step.is_finally:self.is_fail = Trueelif http_step.assert_fail:# finally 步骤如果是断言失败，才会影响用例结果self.is_fail = Trueelse:# 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 Falsenew_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 = datatry:...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信息并更新urlif 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_bodyif 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 valuevalue = 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 valuevalue = str(value)if value.isnumeric():return float(value)value = ast.literal_eval(value)return float(value)
提取参数：将接口的返回数据按提取规则进行提取并存储到对应的变量空间中。
执行步骤的“后置脚本”。
由执行器提供的工具方法，供用户态通过代理（SysFuncsProxy）进行使用的工具类方法，可以在占位符和脚本代码中使用，如：生成随机数、处理时间等方法。
系统方法库
装载系统方法到方法库，装载后才可以在用户态的脚本和占位符中使用该方法。
def 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 funcreturn decorator
系统方法实现
实现系统方法，需要通过@SysFuncsLib.load_sys_func()注入方法相关信息，以便调用规范校验和快捷模版生成。
案例：转换为时间戳(switch_timestamp)，该方法接收 3 个参数，且都是非缺省值参数。
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 requestsdef 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 = 0for 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:# 未传部分缺省参数时结束循环breaktry:_raw_arg = raw_args[i]if param_rule.param_type is bool:# 布尔类型if _raw_arg.capitalize() == 'True':_arg = Trueelif _raw_arg.capitalize() == 'False':_arg = Falseelif 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]：切片（顺序颠倒）
【页面正使用占位符】
占位符使用案例
案例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]']。
def 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_keysraw_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_keysreturn sub_keys
占位符参数提取
根据提取到的子表达式（sub_keys）组层级匹配获取数据。
def get_var(self, key: str):"""获取变量值"""sub_keys = self.parsing_sub_keys(key)if not sub_keys:return Nonesub_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 vardef get_target_var(cls, sub_keys: list, i_var: any, raw_key: str):"""获取指定空间中的变量值"""is_found = Falsevar = Nonei = 0for i, key in enumerate(sub_keys):is_found = Falseif 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_vari_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 = Trueelse: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] = Noneexcept 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 = Trueelse:i_var = i_var[_i_temps[0]: _i_temps[1]: _i_temps[2]]is_found = Trueelse: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 = Trueelse:raise ParseKeyException(f'表达式与实际变量值不匹配取值时索引越界(list index out of range)`{raw_key}`')if i == len(sub_keys) - 1 and is_found:var = i_varreturn var, is_found
执行器中提供了丰富的内置库（Python 自带内置库、第三方库、平台自研内置库）来满足各种使用场景，可以在任何位置的脚本中导入使用。
自动化执行器是自动化平台老方案中自动化框架的超集，提供框架能力的同时还负责解释执行自动化用例。技术实现上融入了优秀自动化框架的核心理念，也拓展支持了很多实用功能，同时更灵活、更具可塑性、更适配平台化，并支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例，也能满足低门槛、高效率产出自动化用例的需要，还能实现极其复杂的测试场景。在自动化平台转型后的近两年时间里，平台在质量保障、效率提升和使用体验方面效果显著，后续我们也将持续打磨完善执行器和自动化平台。
往期回顾
关注得物技术，每周一、三、五更新技术干货
要是觉得文章对你有帮助的话，欢迎评论转发点赞~
未经得物技术许可严禁转载，否则依法追究法律责任。
“
扫码添加小助手微信
如有任何疑问，或想要了解更多技术资讯，请添加小助手微信：