得物自动化平台执行器设计与实现
目录
一、名词解释
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_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]：切片（顺序颠倒）
【页面正使用占位符】
占位符使用案例
案例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 自带内置库、第三方库、平台自研内置库）来满足各种使用场景，可以在任何位置的脚本中导入使用。
九
总结
自动化执行器是自动化平台老方案中自动化框架的超集，提供框架能力的同时还负责解释执行自动化用例。技术实现上融入了优秀自动化框架的核心理念，也拓展支持了很多实用功能，同时更灵活、更具可塑性、更适配平台化，并支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例，也能满足低门槛、高效率产出自动化用例的需要，还能实现极其复杂的测试场景。在自动化平台转型后的近两年时间里，平台在质量保障、效率提升和使用体验方面效果显著，后续我们也将持续打磨完善执行器和自动化平台。
往期回顾
1. SpEL应用实战｜得物技术
2. 得物大模型平台接入最佳实践
3. R8疑难杂症分析实战 - 类反射篇｜得物技术
4. 解密得物Trace2.0：日PB级数据量下的计算与存储性能优化实战
关注得物技术，每周一、三、五更新技术干货
要是觉得文章对你有帮助的话，欢迎评论转发点赞~
未经得物技术许可严禁转载，否则依法追究法律责任。
“
扫码添加小助手微信
如有任何疑问，或想要了解更多技术资讯，请添加小助手微信：