工具层handle_replace
工具层handle_replace
1. 删除换行符和空格
自定义方法名 所用模块/函数 解决的问题 __handle_str
for循环
replace函数
数据清洗:清理请求参数中的 \n
和空格,确保JSON格式合法,避免解析错误
2. 提取需替换的参数名
自定义方法名 所用模块/函数 解决的问题 __get_replace_keys
re
模块:Python的正则表达式操作模块findall
函数:返回字符串中所有匹配正则的子串(列表形式)正则表达式#(\w.+?)#:
匹配任意字符至少一次,但遇到第一个结束标记(如#
)就停止,避免跨越多余内容;
\w
→ 匹配字母/数字/下划线.+?
→ 非贪婪匹配任意字符(至少一个)通过正则 #(\w.+?)#
找出用#包裹起来的参数名并提取出来识别为动态参数(如#user_name#
)
3. 动态设置类属性值
自定义方法名 条件语句 所用模块/函数 __get_data_and_set_attribute
if key in user_info:
取user_info[key]
的值,并设置为HandleAttr
类的属性setattr():
Python 内置函数,动态设置对象的属性值elif key == "sessionUUID":
生成形如a1b2c3d4-1234-5678-90ab-cdef12345678
的UUID并存入类属性uuid.uuid4():uuid
标准模块,生成全局唯一标识符(UUID 格式字符串)elif key == "partyCode" or "bizPayNo":
生成形如1620000000000
的毫秒级时间戳并存入类属性(若属性不存在)time.time():time
标准模块,生成唯一业务编码(如partyCode
、bizPayNo
),避免订单号或流水号重复elif key == "mobile":
生成形如13812345678
的手机号并存入类属性(若属性不存在)handlePhone.get_phone():
自定义类方法,生成未注册的手机号
4. SQL参数替换
逻辑步骤 自定义方法名 所用模块/函数 解决的问题 replace_sql
re.findall()
getattr()
替换SQL中的动态参数(如 #partyCode#
),生成可执行SQL语句5. 执行SQL并注入属性 __execute_sql_and_setattr
mysql.get_data_dict()
setattr()
执行预处理后的SQL,将查询结果映射到类属性(如用户ID、订单号),供后续接口请求使用 6. 整合属性设置流程 __set_attribute
ast.literal_eval()
self.replace_sql()
self.__execute_sql_and_setattr
统一处理属性来源:
- 配置数据
- SQL查询结果
- 动态生成值,确保参数替换前数据就绪7. 主流程:生成最终请求体 replace_data
ast.literal_eval()
self.__handle_str
self.__set_attribute
入口方法:
1. 清洗原始数据 → 2. 提取参数名 → 3. 设置属性 → 4. 替换参数 → 5. 生成JSON请求体
{ #时间戳 "t": "", #用户名 "principal":"#user_name#", #密码 "credentials":"123456a", #类似于access-token "sessionUUID": session_uuid, #验证码 "imageCode":"lemon" } 逻辑: 1、传入请求参数data,excel中读取到的请求参数 2、删除请求参数中的换行符和空格 3、获取需要替换的参数名称 4、通过参数名称获取到参数的值,获取到值统一设置为类属性 5、将参数的值替换掉参数名称和#号 """ import re #正则表达式会用 import ast import uuid import time from tools.handle_path import data_dir from tools.handle_excel import HandleExcel from conf.setting import user_info from tools.handle_attribute import HandleAttr from tools.handle_phone import HandlePhone from tools.handle_db import mysql from tools.handle_logs import myLog class HandleReplace: def __init__(self): self.handle_phone = HandlePhone() # 删除换行符和空格键 # data(str): excel中获取的请求参数 # return data:去掉空格后的请求参数 def __handle_str(self,data:str): for str_data in ["\n"," "]: data = data.replace(str_data,"") return data # 获取需要替换参数的名称 def __get_replace_keys(self,data:str): #"#(\w.+?)#"正则表达式 key_list = re.findall("#(\w.+?)#",data) return key_list # 根据数据的来源,获取数据,设置为类属性 def __get_data_and_set_attribute(self,key:str,user_info): if key in user_info: # 取出来设置为类属性 setattr(HandleAttr,key,str(user_info[key])) elif key == "sessionUUID": # 特殊处理,通过脚本生成sessionUUID setattr(HandleAttr,key,str(uuid.uuid4())) # 添加产品的唯一字段partyCode,单独处理 elif key == "partyCode" or key == "bizPayNo": # 添加产品数据库断言的时候,partyCode要使用参数替换时候的partyCode,不能在sql语句替换的时候再去生产partyCode if hasattr(HandleAttr,key): pass else: setattr(HandleAttr,key,str(int(time.time()*1000))) # 如果是手机号(注册用户),单独生产未注册的手机号 elif key == "mobile": if hasattr(HandleAttr, key): pass else: print("生成为注册的手机号") phone = self.handle_phone.get_phone() setattr(HandleAttr,key,str(phone)) #替换sql语句中的参数 def replace_sql(self,sql,assert_db_info): # 获取需要替换的参数的名称,返回类型list key_list = self.__get_replace_keys(data=sql) if len(key_list)>0: # 获取key的值,然后设置为类属性 for key in key_list: self.__get_data_and_set_attribute(key=key,user_info=assert_db_info) #从类属性中查询到key的值,然后替换 for key in key_list: sql = sql.replace(f"#{key}#", str(getattr(HandleAttr, key))) return sql else: print("不需要替换sql") return sql # 执行sql语句,将sql语句查询出来的key,val设置为类属性的key,val def __execute_sql_and_setattr(self,sql): # 调sql执行方法,获取到数据 # 拿到数据并设置为属性 result = mysql.get_data_dict(sql=sql) for dict_data in result: for key,value in dict_data.items(): setattr(HandleAttr,key,str(value)) # 调用数据来源设置属性方法,设置类属性 def __set_attribute(self,key_list,setup_sql,setup_sql_info): """ : param setup_sql (str):接口执行之前,需要从通过sql查询来获取替换数据,可以兼容多条[sql1,sql2] :param key_list: ["user_name","session_uuid"] :return: """ if setup_sql: for sql in ast.literal_eval(setup_sql): # 替换sql语句中需要替换的参数 new_sql = self.replace_sql(sql=sql,assert_db_info=setup_sql_info) # 执行sql语句,并设置为属性 self.__execute_sql_and_setattr(sql=new_sql) # 只对excel请求参数进行属性设置 for key in key_list: #key的值的来源配置文件 #通过脚本生成的数据 #响应结果 #数据库 self.__get_data_and_set_attribute(key=key,user_info=user_info) # 获取参数,不同的数据,来源是不是不一样,然后设置为类属性 # data (str): excel中读取到的请求参数 def replace_data(self,data,setup_sql,setup_sql_info): try: myLog.info(msg="replace_data参数替换入参:") myLog.info(msg=f"data:{data},{type(data)}") myLog.info(msg=f"setup_sql:{setup_sql},{type(setup_sql)}") myLog.info(msg=f"setup_sql_info:{setup_sql_info},{type(setup_sql_info)}") if data: # 删除换行符和空格键,返回类型str new_data = self.__handle_str(data=data) # 获取需要替换的参数的名称,返回类型list key_list = self.__get_replace_keys(data=new_data) myLog.info(msg=f"需要替换的参数名称:{key_list}") if len(key_list)>0: # 当 len(key_list)>0 说明 参数里面需要替换数据 # 获取参数,并设置为类属性,无返回值 self.__set_attribute(key_list=key_list,setup_sql=setup_sql,setup_sql_info=setup_sql_info) # 从类属性中通过key_list里面的key获取值,然后替换请求参数 for key in key_list: new_data = new_data.replace(f"#{key}#",str(getattr(HandleAttr,key))) new_data = ast.literal_eval(new_data) new_data["t"] = int(time.time()*1000) myLog.info(msg=f"参数替换完成后返回数据:{new_data},{type(new_data)}") return new_data else: new_data = ast.literal_eval(new_data) new_data["t"] = int(time.time() * 1000) myLog.info(msg=f"不需要参数替换,返回数据:{new_data},{type(new_data)}") return new_data else: myLog.info(msg="没有请求参数,不需要做任何事情,返回空的dict") return {} except Exception as e: myLog.error(msg=f"replace_data参数替换方法执行报错,错误原因{e}") myLog.exception(e) #将错误信息收集到日志文件 raise TypeError #手动抛出异常给框架 if __name__ == '__main__': print(case_list[0]) cl = HandleReplace() cl.replace_data(data=case_list[0]["data"])
未待完续!