更新日志
2023.2.7：修复json[key]为object类型时（allpairs解析错误），参数生成格式错误的问题
2023.2.7：将写入文件的策略更改为分批写入，避免在大量数据下写入时造成数据丢失（openpyxl的坑）
目前支持的功能
1.基于不同的参数类型（str、int、float、bool、list、tuple、dict、set、None、date、datetime、time、timestamp）自动生成正交测试用例
2.可以自定义参数长度与内容，方便覆盖边界值测试
3.提供自定义过滤参数组合的入口，方便自定义
实现思路
1.抓取api信息（目前公司用的swagger）： https://www.cnblogs.com/qtclm/p/17049176.html ，uri、method、params、response，解析完成后写入excle
2.读取抓取完毕的api信息，处理为allpairs所需要的ordereddict
3.调用allpairs工具生成测试用例
4.解析allpairs生成的测试用例（输出为字符串），并处理为dict
5.处理完毕后写入excel
后期优化
1.根据接口响应实现自动断言
2.增加其他接口平台的api抓取(openapi、eolink等)
3.增加其他自动化生成用例的方法（有效、无效等价类，流量回放等）
4.集成到平台
遇到的问题
1.allpairs只支持两个以上的参数生成，因为参数只有一个时，需要自行处理
2.参数为json嵌套时，allpairs输出的参数需要额外特殊处理，这块也是最麻烦的地方
3.可变类型在循环过程中尽量使用深拷贝对象，避免循环运行中被意外修改
4.类变量、实例变量需要合理运用，例如循环时需要合理的进行初始化
使用说明
1.init_enum_data：生成各个数据类型枚举的方法，可以根据实际需要修改
2.dict_parse_generator：递归解析json
3.api_prams_convert：将dict所有的key对应的值根据类型推算为定义的枚举值，并完成动态参数替换
4.generate_all_cases（核心），生成正交测试用例
5.write_params_filed_detail_to_excle（单个params）：输出参数明细到excle（不仅可以输出接口测试用例，文字测试用例也可）
6.write_cases_obj_to_excle（单条case）：输出测试用例到excle
7.batch_write_params_filed_detail_to_excle：将所有params输出到excle，基于接口生成sheet
8.batch_write_cases_obj_to_excle：将所有测试用例输出到excle，基于接口模块生成sheet
生成后的效果
batch_write_params_filed_detail_to_excle：
batch_write_cases_obj_to_excle：
# _*_ coding: UTF-8 _*_
"""
@project -> file : city-test -> rr
@Author : qinmin.vendor
@Date : 2023/1/29 19:44
@Desc : 自动生成接口测试用例：支持正交实验，等价类，边界值
"""
import copy
import json
import os
import random
import re
import sys
from utils.operation_datas import operationExcel
import math
from utils.time_util import timeUtil
from utils.wrapper_util import exec_time_wrapper
from faker import Faker
'''faker使用教程：https://blog.csdn.net/qq_42412061/article/details/122997802'''
class autoGenrateApiCaseAllpairspy(object):
time_util = timeUtil()
CONST_TYPE_ENUM=[]
CONST_TYPE_ENUM_COPY=[]
is_array = False # 判断参数传入的参数是不是list
params_key_list = [] # 判断参数的key对应的value是不是list类型，如果是，就将key存进此对象
excel_path=os.path.join(time_util.get_project_rootpath(),"test_data","interface")
excel_name_params_field_detail = 'auto_genrate_api_case_allpairspy_params_field_detail.xlsx'
excel_name_params_obj = 'auto_genrate_api_case_allpairspy_params_obj.xlsx'
excel_name_cases_data = 'maint-apiv2-api-docs.xlsx'
# 递归解析json，原文：https://blog.csdn.net/qq_45662588/article/details/122265447
read_excel_case_field_desc = {'case': 0, 'uri': 1, 'method': 2, "params_path": 3, "params_body": 5,
"response_body": 7, "skip": 9}
# 数据处理前的case表头，写入excel中最终会删除params_path,且params_body替换为params && response_body替换为expected
write_excel_case_field_desc = {'case': 0, 'uri': 1, 'method': 2, 'params_path': 3, 'params_body': 4,
'response_body': 5, 'skip': 6}
# 存放case_info（基于api对应的模块）是否写入首行数据,此参数无需重置
write_case_info_first_line_sheet_names = []
# 存放params(基于api)是否写入首行数据,此参数无需重置
write_params_first_line_sheet_names = []
# 断言追加http状态响应码
expected_extra_info={"http_status_code":200}
# 定义excle分配保存条数，避免一次性写入太多数据造成数据丢失
batch_save_num=10
@classmethod
# @exec_time_wrapper(round_num=10,module_obj=__file__,class_obj=sys._getframe().f_code.co_name,is_send_email=False)
def init_enum_data(cls):
'''初始化各个类型所需数据'''
fake = Faker(locale="zh_cn") # 指定语言为中文
CONST_MAX_STRING_LENGTH = 100
CONST_MIN_STRING_LENGTH = 1
# 生成字符串相关的数据
# special_chars是否包含特殊字符;digits是否包含数字;upper_case是否包含大写字母,lower_case是否包含小写字母
special_character = fake.password(length=8, special_chars=True, digits=False, upper_case=False,
lower_case=False) # 生成随机特殊字符
random_str = fake.paragraph() # 生成随机字符
random_str_max = random_str * math.ceil(CONST_MAX_STRING_LENGTH / len(random_str))
phone_number = fake.phone_number() # 生成随机电话号码
# 生成时间相关的数据
random_date = fake.date(pattern="%Y-%m-%d", end_datetime=None) # 生成随机日期
random_start_datetime = cls.time_util.adjust_time(num=-30)
random_end_datetime = cls.time_util.adjust_time(num=0)
random_start_timestamp_millisecond = cls.time_util.adjust_time(num=-30, is_timestamp=True, millisecond=True)
random_end_timestamp_millisecond = cls.time_util.adjust_time(num=0, is_timestamp=True, millisecond=True)
random_start_timestamp = cls.time_util.adjust_time(num=-30, is_timestamp=True, millisecond=False)
random_end_timestamp = cls.time_util.adjust_time(num=0, is_timestamp=True, millisecond=False)
# 生成int与float相关的数据
random_int = random.randint(1, 100)
random_float = round(random.random(), 6) + random_int # 生成随机字符
random_int_max = str(random_int) * math.ceil(CONST_MAX_STRING_LENGTH / len(str(random_int)))
random_float_max = str(str(random_float) *
math.ceil(CONST_MAX_STRING_LENGTH / len(str(random_float).replace('.', '')))).\
replace('.','') + "." + str(random_int)
STRING_ENUM = ["", None, random_str, special_character,
random_str_max[:CONST_MAX_STRING_LENGTH], random_str[:CONST_MIN_STRING_LENGTH],
random_start_datetime, random_end_datetime, random_date]
INT_FLOAT_ENUM = [0, None, special_character, random_int, random_float,
int(random_int_max[:CONST_MAX_STRING_LENGTH]),phone_number,
int(str(random_int)[:CONST_MIN_STRING_LENGTH]), int(random_float_max[:CONST_MAX_STRING_LENGTH]),
int(str(random_float)[:CONST_MIN_STRING_LENGTH]), random_start_timestamp,
random_start_timestamp_millisecond, random_end_timestamp, random_end_timestamp_millisecond]
LIST_ENUM=[[], None, [{}],[random_int],[random_str]]
BOOL_ENUM=[None, False, True,"true","false"]
OBJECT_ENUM=[None,{}]
NULL_ENUM=[None,"null"]
# 定义接口字段生成的类型与枚举
cls.CONST_TYPE_ENUM = {
"str": STRING_ENUM,"string": STRING_ENUM,
"int": INT_FLOAT_ENUM,"integer": INT_FLOAT_ENUM,"number": INT_FLOAT_ENUM,
"NONE": NULL_ENUM,"null": NULL_ENUM,
'bool': BOOL_ENUM,"boolean": BOOL_ENUM,
'object': OBJECT_ENUM,"list": LIST_ENUM,'array': LIST_ENUM,
}
cls.CONST_TYPE_ENUM_COPY = copy.deepcopy(cls.CONST_TYPE_ENUM)
@classmethod
def read_all_cases_data(cls):
cases_data_obj=operationExcel(excel_path=cls.excel_path,excel_name=cls.excel_name_cases_data,is_read=True,data_only=False)
lines=cases_data_obj.get_lines()
case_list=[]
for lindex,i in enumerate(range(lines)):
values=cases_data_obj.get_row_data(i+1)
if values[0]:
out_values=[]
for index,i in enumerate(cls.read_excel_case_field_desc):
if lindex!=0 and index==cls.write_excel_case_field_desc['response_body']:
expected_info=json.loads(values[cls.read_excel_case_field_desc[i]])
expected_info.update(cls.expected_extra_info)
expected_info['api_response_code']=expected_info['code'] if expected_info.get('code') else 0
expected_info['code']=expected_info.pop('http_status_code')
values[cls.read_excel_case_field_desc[i]]=json.dumps(expected_info,ensure_ascii=False)
out_values.append(values[cls.read_excel_case_field_desc[i]])
case_list.append(out_values)
return case_list
@classmethod
def dict_parse_generator(cls,params,pre=None):
'''递归解析json，如果key为对应的值是list且list中的元素不是dict，则不返回'''
pre = pre[:] if pre else [] # 纪录value对应的key信息
# print("params:",params)
if isinstance(params,list) :
params=params[0]
if isinstance(params, dict):
for key, value in params.items():
if isinstance(value, dict):
if len(value) > 0:
for d in cls.dict_parse_generator(value, pre + [key]):
yield d
else:
yield pre + [key, '{}']
elif isinstance(value, (list, tuple)):
if len(value) > 0:
for index, v in enumerate(value):
# 纪录list的下标与key，方便动态提取
# print("index",index)
for d in cls.dict_parse_generator(v, pre + [key, str(index)]):
yield d
else:
yield pre + [key, '[]'] if isinstance(value,list) else pre + [key,'()']
else:
yield pre + [key, value]
@classmethod
def api_prams_convert(cls, params_info):
'''根据参数与参数类型转换为预期定义的数据类型定义的枚举'''
def json_key_is_object_convert_list(params_info):
'''#将json[key]-> object 转换为json[key]-> list[object]，由于allpairs无法解析，所以这里手动进行转换'''
for i in params_info:
if isinstance(params_info[i],dict):
cls.time_util.key_is_object_convert_list.append(i)
params_info[i]=[params_info[i]]
def get_params_field_type(params_field_value):
if str(params_field_value).startswith('"') and str(params_field_value).endswith('"'):
_type = 'string' or 'str'
elif str(params_field_value).startswith('{') and str(params_field_value).endswith("}"):
_type = 'object' or 'dict'
elif str(params_field_value).startswith('[') and str(params_field_value).endswith(']'):
_type = 'array' or 'list'
elif str(params_field_value) in ('0', '0.0'):
_type = 'number' or 'int'
elif str(params_field_value) in ("true", 'false', 'True', 'False'):
_type = 'boolean' or 'bool'
elif str(params_field_value) in ('None', 'null', 'nil'):
_type = 'null' or 'None'
else:
_type = 'string' or 'str'
return _type
def replace_params_values(params_info):
'''替换参数所有key的内容'''
parse_params_info = list(cls.dict_parse_generator(params_info))
parse_params_info_keys=[i[0] for i in parse_params_info]
# 将没有输出的key添加到字典
for params_info_key in params_info:
if params_info_key not in parse_params_info_keys:
parse_params_info.append([params_info_key,params_info[params_info_key]])
for i in range(len(parse_params_info)):
if parse_params_info[i] :
params_key = parse_params_info[i][:-1]
# 组装json提取表达式
params_extract_str = "']['".join(params_key)
params_extract_str = "['" + params_extract_str + "']"
params_extract_str_re = re.search("\['\d+'\]", params_extract_str)
if params_extract_str_re:
# 去除list索引下标的字符串，处理为eval能够识别的int字符串
params_extract_str_re = params_extract_str_re.group()
params_extract_str = params_extract_str.replace(params_extract_str_re,
params_extract_str_re.replace("'",'').replace('"',""))
# 获取key内容
params_field_value=eval(f"params_info{params_extract_str}")
params_type=get_params_field_type(params_field_value=params_field_value)
# # 将不为空得参数值且不在枚举定义范围内的数据添加到枚举定义中，避免影响原有数据且用于完善参数覆盖情况
if params_field_value and (params_field_value not in cls.CONST_TYPE_ENUM[params_type]):
cls.CONST_TYPE_ENUM[params_type].append(params_field_value)
#动态执行代码：替换key的内容
exec(f"params_info{params_extract_str}={cls.CONST_TYPE_ENUM[params_type]}")
def append_params_key_is_object_to_params_key_list(params_info):
'''追加参数key对应的值是object类型的key到指定list中'''
if isinstance(params_info,list):
params_info=params_info[0]
for i in params_info:
if isinstance(params_info[i],list) and isinstance(params_info[i][0],dict)\
and (i not in cls.params_key_list):
cls.params_key_list.append(i)
if not params_info:
return params_info
if isinstance(params_info, (list, dict)):
if isinstance(params_info, list):
if params_info:
params_info = params_info[0]
cls.is_array = True
json_key_is_object_convert_list(params_info)
replace_params_values(params_info)
append_params_key_is_object_to_params_key_list(params_info)
return params_info
@classmethod
def parse_uri_params_path_query(cls,uri: str, params_path: dict):
'''解析url中的params_path与params_query参数'''
if not params_path:
return {},{}
uri_split = uri.split('/')
uri_params = [i for i in uri_split if i.startswith('{') and i.endswith('}')]
params_path_keys = list(params_path.keys())
uri_params_path_dict = {} # 存储：uri中的参数是key的情况
uri_params_query_dict = {} # 存储：uri中的参数不是key的情况，代表是query参数
for params_path_key in params_path_keys:
if "{" + params_path_key + "}" in uri_params:
uri_params_path_dict[params_path_key]=params_path[params_path_key]
else:
uri_params_query_dict[params_path_key]=params_path[params_path_key]
return uri_params_path_dict,uri_params_query_dict
@classmethod
def join_request_uri(cls,uri,path_dict,query_dict):
def out_path_replace_uri(uri,dict_obj):
'''uri拼接替换后的path参数'''
if dict_obj:
for i in dict_obj:
uri = uri.replace("{" + i + "}",str(dict_obj[i]) )
return uri
return uri
def out_query_replace_uri(uri,dict_obj):
'''uri拼接替换后的query参数'''
if dict_obj:
uri += "?"
for i in dict_obj:
uri += f"{i}={str(dict_obj[i])}&"
uri = uri[:-1] if uri[-1] == "&" else uri
return uri
if not (path_dict or query_dict):
return uri
uri=out_path_replace_uri(uri=uri,dict_obj=path_dict)
uri=out_query_replace_uri(uri=uri,dict_obj=query_dict)
uri = cls.time_util.pyobject_to_json_str(uri)
return uri
@classmethod
def out_target_case(cls,src_obj_index, params_all_obj):
'''根据index返回符合条件的case'''
if not params_all_obj:
return {}
_l = len(params_all_obj) - 1
if src_obj_index > _l:
target_case = params_all_obj[random.randint(0, _l)]
else:
target_case = params_all_obj[src_obj_index]
return target_case
@classmethod
# @exec_time_wrapper(round_num=10,module_obj=__file__,class_obj=sys._getframe().f_code.co_name,is_send_email=False)
def generate_all_cases(cls, ordered_dict_obj,params_info):
def convert_allpairs_cases_to_cases(params_keys,ordered_dict_obj,cases_obj):
'''将allparis输出的case字符串转换为dict输出'''
if len(params_keys)>=2:
cls.time_util.append_params_key_isobject_to_extra_cases(case_params_key_obj=ordered_dict_obj,
params_keys=params_keys,extra_cases=cls.time_util.extra_cases,filter_func=cls.time_util.all_pairs_filter_func,
cases_obj=cases_obj,is_append_extra_cases=False,params_key="")
else:
'''params只有一个key时,直接组装字典输出（allpairs这种情况处理会输出空）'''
cases = ordered_dict_obj
for case in cases:
for case_key in cases[case]:
cases_obj.append({case: case_key})
def append_extra_case_to_case(cases_obj,extra_cases,params_key_list):
# 将extra_case数据追加到参数中
for index,out_case in enumerate(cases_obj):
for params_key in params_key_list:
update_cases=[i for i in extra_cases for o in i if o == params_key]
extra_case=cls.out_target_case(src_obj_index=index,params_all_obj=update_cases)
out_case.update(extra_case)
def return_out_cases(params_info,ordered_dict_obj):
'''输出最终转换完成的测试用例集合'''
if isinstance(params_info,list):
params_info=params_info[0]
params_keys=[i for i in params_info]
out_cases = []
'''将allparis输出的case字符串转换为dict输出'''
convert_allpairs_cases_to_cases(params_keys=params_keys,ordered_dict_obj=ordered_dict_obj,cases_obj=out_cases)
'''将extra_case数据追加到参数中'''
append_extra_case_to_case(cases_obj=out_cases,extra_cases=cls.time_util.extra_cases,params_key_list=cls.params_key_list)
return out_cases
if not (params_info or ordered_dict_obj):
return []
return return_out_cases(params_info=params_info,ordered_dict_obj=ordered_dict_obj)
@classmethod
def rm_old_excel(cls, excel_path='./', excel_name=''):
if os.path.exists(os.path.join(excel_path, excel_name)):
os.remove(os.path.join(excel_path, excel_name))
@classmethod
def reset_params_key_list_and_extra_cases(cls):
'''每个api的参数不同，在写入后需要调用此方法重置，避免元素被重复使用'''
cls.params_key_list=[]
cls.time_util.extra_cases=[]
cls.time_util.extra_cases_hash_lists=[]
cls.time_util.key_is_object_convert_list=[]
# 还原默认的枚举定义，避免数据一直递增
cls.CONST_TYPE_ENUM=cls.CONST_TYPE_ENUM_COPY
cls.is_array=False
@classmethod
def write_params_filed_detail_to_excel(cls, params,ex_obj:operationExcel=None):
'''将参数字段明细写入到excel'''
ex_obj_cp=copy.deepcopy(ex_obj)
if ex_obj is None:
cls.rm_old_excel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_field_detail)
ex_obj = operationExcel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_field_detail,is_read=False,data_only=False)
if not isinstance(ex_obj,operationExcel):
raise Exception(f"传入的对象预期是:{type(operationExcel)},实际为:{type(ex_obj)}")
# print("params:",params)
if isinstance(params,dict):
params=[params]
ordered_dict_obj = cls.api_prams_convert(params)
params_all = cls.generate_all_cases(ordered_dict_obj, params)
for case in params_all:
# 写入首行数据
first_line=list(case.keys())
hash_first_line_str=cls.time_util.encry_main(str(first_line))
if not hash_first_line_str in cls.write_params_first_line_sheet_names:
ex_obj.write_values(first_line)
cls.write_params_first_line_sheet_names.append(hash_first_line_str)
for case_key in case:
# 如果key存在与list中，说明参数是list,将object转换为array
if case_key in cls.params_key_list:
case[case_key] = [case[case_key]]
ex_obj.write_values([str(i) for i in case.values()])
# 重置参数状态
cls.reset_params_key_list_and_extra_cases()
if ex_obj_cp is None:
ex_obj.write_values(list(params_all[0].keys()))
ex_obj.save_workbook()
@classmethod
def write_cases_obj_to_excel(cls, case_info, params_path,params_body,ex_obj:operationExcel=None):
def write_cases_data_main(case_info,params_all_cases,is_body=False):
'''
将生成完毕的测试用例写入到excel，基础方法（将测试用例明细写入到excel）
Args:
case_info: 原始的测试用例信息
params_all_cases: 使用正交生成后的所有测试用例集合
is_body: 用于区分请求参数是否是body，如果不是bodu，则设置params_bodu为{},因为path与query类型不需要body请求，参数直接拼接到url上了
Returns:
'''
for index, case in enumerate(params_all_cases):
# print("params_all_cases:",id(params_all_cases))
path_dict = cls.out_target_case(src_obj_index=index,
params_all_obj=params_all_path) if params_all_path else {}
query_dict = cls.out_target_case(src_obj_index=index,
params_all_obj=params_all_query) if params_all_query else {}
# 拼接uri
uri = cls.join_request_uri(uri=uri_copy, path_dict=path_dict, query_dict=query_dict)
case_identifying_str = '-接口异常测试用例(基于正交实验自动生成)'
case_name = case_info[cls.write_excel_case_field_desc['case']]
# 拼接测试用例名称
if case_identifying_str in case_name:
case_name = case_name[:case_name.find(case_identifying_str)]
case_name = case_name + case_identifying_str + '-' + str(index + 1)
case_info[cls.write_excel_case_field_desc['case']] = case_name
case_info[cls.write_excel_case_field_desc['uri']] = uri
for case_key in case:
# 如果key存在与list中，说明参数是list,将object转换为array
if (case_key in cls.params_key_list) and (case_key not in cls.time_util.key_is_object_convert_list) :
case[case_key] = [case[case_key]]
# 判断最外层的参数是不是list，是list,将object转换为array
if cls.is_array:
case_info[cls.write_excel_case_field_desc['params_body']] = json.dumps([case], ensure_ascii=False)
else:
case_info[cls.write_excel_case_field_desc['params_body']] = json.dumps(case, ensure_ascii=False)
if not is_body:
# 从params_body中删除params_path¶ms_query对应的key
case_info[cls.write_excel_case_field_desc['params_body']] = json.dumps({})
ex_obj.write_values(case_info)
# ex_obj_cp=copy.deepcopy(ex_obj)
if ex_obj is None:
cls.rm_old_excel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_obj)
ex_obj = operationExcel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_obj,data_only=False,is_read=False)
if not isinstance(ex_obj,operationExcel):
raise Exception(f"传入的对象预期是:{type(operationExcel)},实际为:{type(ex_obj)}")
uri=case_info[cls.write_excel_case_field_desc['uri']]
uri_copy=copy.deepcopy(uri)
# 根据params_path对象区分uri_path参数、uri_query参数
uri_params_path_dict,uri_params_query_dict=cls.parse_uri_params_path_query(uri=uri_copy,params_path=params_path)
# 输出path与body的ordered_dict对象
ordered_dict_obj_path = cls.api_prams_convert(uri_params_path_dict)
ordered_dict_obj_query = cls.api_prams_convert(uri_params_query_dict)
# 输出path与body所有的正交测试用例
params_all_path = cls.generate_all_cases(ordered_dict_obj_path, uri_params_path_dict)
params_all_query = cls.generate_all_cases(ordered_dict_obj_query, uri_params_query_dict)
ordered_dict_obj_body = cls.api_prams_convert(params_body)
# print("ordered_dict_obj_body:",ordered_dict_obj_body)
params_all_body = cls.generate_all_cases(ordered_dict_obj_body, params_body)
if params_all_body:
write_cases_data_main(case_info=case_info,params_all_cases=params_all_body,is_body=True)
elif params_all_path:
write_cases_data_main(case_info=case_info, params_all_cases=params_all_path, is_body=False)
elif params_all_query:
write_cases_data_main(case_info=case_info, params_all_cases=params_all_query, is_body=False)
# 重置参数状态
cls.reset_params_key_list_and_extra_cases()
@classmethod
@exec_time_wrapper(round_num=10,module_obj=__file__,class_obj=sys._getframe().f_code.co_name,is_send_email=True)
def batch_write_params_filed_detail_to_excel(cls,case_list=None):
if case_list is None:
case_list=cls.read_all_cases_data()
# 初始化数据
cls.init_enum_data()
cls.rm_old_excel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_field_detail)
ex_obj = operationExcel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_field_detail,is_read=False,data_only=False)
if isinstance(case_list,dict):
case_list=[case_list]
for index,params in case_list[1:]:
# params是一个可变类型，程序执行中共用了此对象，这里需要传一个深拷贝对象给他，避免执行过程中params被替换
params=copy.deepcopy(params)
params_body=params[cls.write_excel_case_field_desc['params_body']]
# 根据case名称创建sheet，并切换到对应的sheet
sheet_name=params[cls.write_excel_case_field_desc['case']][:31]
sheet_name=cls.time_util.check_filename(sheet_name,priority_matching_chars=[':','/','\\','?','*','[',']'])
# print("sheet_name:",sheet_name)
if not sheet_name in ex_obj.get_all_sheet_name():
ex_obj.create_sheet(sheet_name,is_save=True)
ex_obj.data=ex_obj.get_data_for_sheet_name(sheet_name)
if not isinstance(params_body, dict):
params_body = json.loads(params_body)
if isinstance(params_body,list):
params_body=params_body[0]
cls.write_params_filed_detail_to_excel(params=params_body,ex_obj=ex_obj)
if index>0 and index%cls.batch_save_num==0:
ex_obj.save_workbook()
# 删除默认创建的sheet
ex_obj.delete_sheet(sheet_name='Sheet')
ex_obj.save_workbook()
@classmethod
@exec_time_wrapper(round_num=10,module_obj=__file__,class_obj=sys._getframe().f_code.co_name,is_send_email=True)
def batch_write_cases_obj_to_excel(cls,case_info=None,case_list=None):
if case_list is None:
case_list=cls.read_all_cases_data()
if isinstance(case_list, dict):
case_list = [case_list]
case_info_first_line=case_info
if case_info is None:
case_info_first_line=list(cls.write_excel_case_field_desc.keys())
# 删除params_path, 且params_body替换为params & & response_body替换为expected
case_info_first_line[cls.write_excel_case_field_desc['params_body']]='params'
case_info_first_line[cls.write_excel_case_field_desc['response_body']]='expected'
# 初始化数据
cls.init_enum_data()
# 删除旧的文件
cls.rm_old_excel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_obj)
ex_obj = operationExcel(excel_path=cls.excel_path, excel_name=cls.excel_name_params_obj,is_read=False,data_only=True)
for index,case in enumerate(case_list[1:]):
# params是一个可变类型，程序执行中共用了此对象，这里需要传一个深拷贝对象给他，避免执行过程中params被替换
case_info=copy.deepcopy(case)
# if case_info[1] in (
# # '/v1/work-order/{tenantId}/{aggregateId}/repair-info',
# # # # # # '/v1/wx/user/logout',
# # # # # '/v1/wx/work-order/{tenantId}/{userId}/my-work-orders',
# # # '/v1/work-order/tenant/{tenantId}',
# '/v1/maintenance/map/breakage-list',
# '/v1/maintenance/map/breakage-list-page',
# # '/v1/stakes/section/{sectionId}/number/{stakeSerialNumber}',
# # # '/v1/maintenance/search/select-explicit',
# '/v1/collector/find-aggregate',
# ):
params_body=case_info[cls.write_excel_case_field_desc['params_body']]
params_path=case_info[cls.write_excel_case_field_desc['params_path']]
# 根据case模块名称创建sheet，并切换到对应的sheet
sheet_name=case_info[cls.write_excel_case_field_desc['case']].split('_')[0]
sheet_name=cls.time_util.check_filename(sheet_name,priority_matching_chars=[':','/','\\','?','*','[',']'])
# if sheet_name=='获取字典接口':
if not sheet_name in ex_obj.get_all_sheet_name():
ex_obj.create_sheet(sheet_name,is_save=True)
ex_obj.data=ex_obj.get_data_for_sheet_name(sheet_name)
if not sheet_name in cls.write_case_info_first_line_sheet_names:
ex_obj.write_values(case_info_first_line)
cls.write_case_info_first_line_sheet_names.append(sheet_name)
if not isinstance(params_body,dict):
params_body=json.loads(params_body)
if not isinstance(params_path,dict):
params_path=json.loads(params_path)
if params_body in ({},[],[{}],None) and params_path in ({},[],[{}],None) :
case_info[cls.write_excel_case_field_desc['params_body']]=json.dumps(params_body)
case_info[cls.write_excel_case_field_desc['params_path']] = json.dumps(params_path)
ex_obj.write_values(case_info)
else:
cls.write_cases_obj_to_excel(case_info=case_info,params_path=params_path,params_body=params_body,ex_obj=ex_obj)
if index>0 and index%cls.batch_save_num==0:
ex_obj.save_workbook()
# 删除params_path对应的列
for sheet_name in ex_obj.get_all_sheet_name():
ex_obj.data=ex_obj.get_data_for_sheet_name(sheet_name=sheet_name)
ex_obj.del_ws_cols(cls.write_excel_case_field_desc['params_path']+1,is_save=False)
#删除load_workbook默认创建的sheet
ex_obj.delete_sheet(sheet_name='Sheet')
ex_obj.save_workbook()
if __name__ == "__main__":
params = {"account": "demo", "pwd": "crmeb.com", "key": "533721295cb06314f4bcaacebc28e3bd", "code": "nbcw",
"wxCode": "", 'userinfo': {}, 'age': 0, 'is_vip': False, 'ext': None,
'orders': [{"order": "asc", "fileds": []}], 'orders2': [{"order2": "asc", "fileds2": []}],
"price2":0.05,"ip":""}
#
auto_case = autoGenrateApiCaseAllpairspy()
# print(auto_case.read_all_cases_data())
# # 读取case数据
# print(auto_case.read_all_cases_data())
# auto_case.batch_write_cases_obj_to_excel()
# auto_case.join_uri_params_path('/v1/stakes/section/{sectionId}/number/{stakeSerialNumber}',{"sectionId": 0, "stakeSerialNumber": "","test":123})
# 单个写入
# auto_case.write_params_filed_detail_to_excel(params)
# auto_case.write_cases_obj_to_excel(case_info, params=params_array)
# 批量写入
# auto_case.batch_write_params_filed_detail_to_excel()
# print("case_info:before",case_info)
auto_case.batch_write_cases_obj_to_excel()
# print("case_info:after",case_info)
依赖项
time_util.py
# -*- coding: utf-8 -*-
"""
__author__ = qtclm
File Name： time_util
date： 2020/8/12 16:10
"""
import traceback
from dateutil import parser
import re
import sys
import time
import datetime
import dateutil
from utils.other_util import otherUtil
class timeUtil(otherUtil):
# 把datetime转成字符串
def datetime_to_string(self,dateTime):
return dateTime.strftime("%Y-%m-%d %H:%M:%S")
# 把字符串转成datetime
def string_to_datetime(self,string):
string_index=str(string).find(".")
# print("string_index,%s"%string_index)
if string_index!=-1:
string=string[:string_index]
# print(string)
return datetime.datetime.strptime(string, "%Y-%m-%d %H:%M:%S")
# 把字符串转成时间戳形式
def string_to_timestamp(self,strTime='1970-01-01 00:00:00',isTimestamp_second:bool=False):
'''isTimestamp_second:是否输出时间为秒，默认为false即13位时间戳，true：返回10位时间戳'''
# print("strTime,%s"%strTime)
try:
if strTime=='1970-01-01 00:00:00':
strTime=self.timestamp_to_string(timeStamp=0)
timeStamp=int(time.mktime(self.string_to_datetime(strTime).timetuple()))
if not isTimestamp_second:
timeStamp*=1000
return timeStamp
except Exception as error:
print("数据处理失败，原因为:\n%s" % (error))
def timestamp_to_string(self,timeStamp=0,is_date=False):
# 把时间戳转成字符串形式
'''
Args:
timeStamp: 默认为当前时间
is_date: 是否输出日期
Returns:
'''
try:
if not timeStamp:
timeStamp=self.get_timestamp()
timeStamp = int(timeStamp)
if len(str(timeStamp)) >= 13:
timeStamp /= 1000
if not timeStamp and timeStamp!=0:
timeStamp = time.time()
if is_date:
return time.strftime("%Y-%m-%d", time.localtime(timeStamp))
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timeStamp))
except:
traceback.print_exc()
# 把datetime类型转外时间戳形式
def datetime_to_timestamp(self,dateTime,millisecond=False):
if millisecond:
return int(time.mktime(dateTime.timetuple()))*100
return int(time.mktime(dateTime.timetuple()))
def datetime_to_iso_date(self,dateTime):
return dateutil.parser.parse(dateTime)
# 将iso_date转换为时间字符串，iso_date主要是mongo存储得时间格式
def iso_date_to_string(self,iso_date_in=None,time_diffence=0):
'''time_diffence:时间差，目前只支持指定小时'''
if iso_date_in is None:
iso_date_in = datetime.datetime.now()
return iso_date_in
else:
re_date = re.search('(\d.+\d)', iso_date_in)
if re_date:
re_date = re_date.group()
if ":" in re_date:
# print(re_date)
iso_date_in = dateutil.parser.parse(re_date) # 转换为iso_date为时间字符串
time_out = datetime.datetime.strptime(iso_date_in.strftime('%Y-%m-%d %H:%M:%S'), '%Y-%m-%d %H:%M:%S')
# datetime.timedelta对象代表两个时间之间的时间差,这里需要计算八个小时后得时间，所以指定hours=8，
# 当前也指定其他时间对象day、seconds、microseconds、milliseconds、minutes、hours、weeks、fold等
delta = datetime.timedelta(hours=time_diffence)
iso_date_out = str(time_out + delta)
else:
# print(re_date)
iso_date_out=self.timestamp_to_string(re_date)
return iso_date_out
else:
return iso_date_in
# 处理robo 3t复制出来的mongo查询结果，序列化为list
def mongodata_serialize(self,str_in,space_one='NumberLong\(\W?\d+\)',space_two="\/\*\s?\d+\s?\*\/",space_date='iso_date\(.*\)'):
str_out = str(self.jsonStrToPyobject(str_in))
'''处理NumberLong'''
mongo_numberLong_list = re.findall(space_one, str_out)
if mongo_numberLong_list:
for i in mongo_numberLong_list:
mongo_numberLong = re.search('[-+]?\d+', i)
if mongo_numberLong:
mongo_numberLong = mongo_numberLong.group()
str_out = str_out.replace(i, mongo_numberLong)
else:
str_out=str_out
'''处理iso_date'''
iso_date=re.findall(space_date,str_out)
if iso_date:
for iso_date_in_str in list(set(iso_date)):
# print(iso_date_in_str)
iso_date_out_str=str(self.string_to_datetime(self.iso_date_to_string(iso_date_in=iso_date_in_str))*1000)
str_out=str_out.replace(iso_date_in_str,iso_date_out_str)
else:
str_out=str_out
'''处理集合间的分隔符'''
mongo_separator_list = re.findall(space_two, str_out)
if not mongo_separator_list:
return eval(str_out)
for i in mongo_separator_list:
mongo_separator = re.search('[-+]?\d+', i)
if mongo_separator:
if mongo_separator.group() == '1':
mongo_separator = '['
else:
mongo_separator = ','
str_out = str_out.replace(i, mongo_separator)
str_out += ']'
return eval(str_out)
# 字符串转换为时间戳,支持调整
def string_to_timestamp_adjust(self,dateTimestr=None,type:int=2,num:int=0,isTimestamp_second:bool=False):
time_to_str=self.adjust_time(type=type,num=num,dateTimestr=dateTimestr)
timestamp=self.string_to_timestamp(time_to_str,isTimestamp_second=isTimestamp_second)
return timestamp
# # 字符串转换为时间戳,支持调整
def timestamp_to_string_ajdust(self,dateTimestr=None, type: int = 2, num: int = 0):
timeStamp = self.string_to_timestamp_adjust(type=type, num=num,dateTimestr=dateTimestr)
return self.timestamp_to_string(timeStamp)
# 输出当前时间的13位时间戳
def get_timestamp(self,is_secends=False):
'''
Args:
is_secends: 是否单位输出为秒
Returns:
'''
if is_secends:
current_milli_time = lambda: int(round(time.time()))
else:
current_milli_time = lambda: int(round(time.time()))*1000
# 输出13位时间戳,round:对浮点数进行四舍五入
return str(current_milli_time())
# 时间调整
def adjust_time(self,dateTimestr=None,type:int=2,num:int=0,is_timestamp=False,millisecond=False):
'''days seconds microseconds milliseconds minutes hours weeks fold'''
'''type: 周 天 时 分 秒
isDate: 是否是日期， 默认否'''
if not dateTimestr:
strTime=datetime.datetime.now()#默认取当前时间
else:
strTime=self.string_to_datetime(dateTimestr)
# strTime=dateTimestr
if type == 1:
day = strTime + datetime.timedelta(weeks=num)
elif type==2:
day = strTime+ datetime.timedelta(days=num)
elif type==3:
day = strTime + datetime.timedelta(hours=num)
elif type==4:
day = strTime + datetime.timedelta(minutes=num)
elif type==5:
day = strTime + datetime.timedelta(seconds=num)
else:
print("暂不支持的调整单位")
sys.exit()
# print(day)
# 将当前时间转换后时间戳，然后在将时间戳转换为时间字符串
if is_timestamp:
return self.datetime_to_timestamp(day,millisecond=millisecond)
return self.timestamp_to_string(self.datetime_to_timestamp(day,millisecond=False))
# 生成mongo时间字符
def genrate_mongo_iso_date(self,dateTimestr=None,type:int=2,num:int=0,isDate:bool=False,isMongo=True):
'''
:param dateTimestr: 默认取当前时间，如果传入了时间则取传入的时间
:param type: 调整时间的类型
:param num: 调整时间的数值
:param isDate： 是否输出日期
:param isMongo: 是否输出mongo格式
:return:
'''
'''days seconds microseconds milliseconds minutes hours weeks fold'''
'''type: 周 天 时 分 秒
isDate: 是否是日期， 默认否'''
# 输出当前日期
if not dateTimestr :
temp_datetime= self.string_to_datetime(self.adjust_time(type=type,num=num))
else:
temp_datetime=self.string_to_datetime(self.adjust_time(dateTimestr=dateTimestr,type=type,num=num))
# print("temp_datetime,%s"%temp_datetime)
day = temp_datetime.strftime('%Y-%m-%d')
dateStr = "%sT%s.000Z" % (day,temp_datetime.strftime('%H:%M:%S'))
if isMongo:
if isDate:
iso_date = parser.parse(day)
return iso_date
iso_date = parser.parse(dateStr) # 将time字符串转换为mongo iso_date
return iso_date
else:
if isDate:
return day
return dateStr
if __name__ == "__main__":
du=timeUtil()
print(du.timestamp_to_string(is_date=True))
# print(du.genrate_mongo_iso_date(isDate=True,isMongo=False))
# print(du.adjust_time(is_timestamp=False,millisecond=True))
# print(du.adjust_time(is_timestamp=False,dateTimestr='2023-01-19 16:57:17',millisecond=True,type=5,num=2223))
# print(du.adjust_time(dateTimestr="2021-06-02 11:14:27"))
# print(du.adjust_time(dateTimestr=None))
# print(du.genrate_mongoiso_date(isMongo=False))
# print(du.adjust_time(strTime="2021-06-03 16:32:36",num=2))
# print(du.timestamp_toString_ajdust(strTime="2021-06-03 16:32:36",num=2))
# print(du.string_toTimestamp())
# print(du.timestamp_toString(stamp=0))
# print(du.string_toTimestamp_adjust(num=1))
# print(du.timestamp_toString_ajdust(num=2))
# print(parser.parse("2021-05-26 17:33:55.000003"))
# print(du.datetime_toiso_date("2021-05-26 17:33:55"))
# print(du.adjust_time(type=1))
# print(du.string_toTimestamp_adjust(num=2,isTimestamp_second=True))
other_util.py
# _*_ coding: UTF-8 _*_
"""
@project -> file : city-test -> rr
@Author : qinmin.vendor
@Date : 2023/1/29 19:44
@Desc :
"""
import ast
import hashlib
import os
import re
import sys
from collections import OrderedDict
from inspect import isfunction
from urllib.parse import urljoin,urlparse,urlunparse
from posixpath import normpath
from allpairspy import AllPairs
from utils.wrapper_util import exec_time_wrapper
class otherUtil(object):
# 输入字符串，输出字符串的大小写与首字母大写以及字符串本身
extra_cases = [] #存放额外的case,请求参数中的key嵌套json的情况
extra_cases_hash_lists=[] #存放extra_cases元素的hash值，增加程序判断的效率
key_is_object_convert_list=[] #将json[key]-> object 转换为 json[key]-> list[object]
@classmethod
def custom_valid_combination(cls,row):
"""
自定义过滤条件，返回false的条件不会出现在组合中
"""
n = len(row)
if n > 1:
if row[0] in (None,'',""):
return False
return True
# 指定：自定义过滤函数
all_pairs_filter_func=custom_valid_combination
if not (isfunction(all_pairs_filter_func) or isinstance(all_pairs_filter_func,(classmethod,staticmethod))) :
raise Exception(f"all_pairs_filter_func对象必须是一个方法:{all_pairs_filter_func}")
@classmethod
# @exec_time_wrapper(round_num=10,module_obj=__file__,class_obj=sys._getframe().f_code.co_name,is_send_email=False)
def recursion_remove_char(cls,str_in,space_char,remove_type=2):
'''
Args:
str_in:
space_char: 需要被移除的字符
remove_type: 移除类型，0：去除字符串左边的字符,1：去除右边的字符，2：去除左右两边的字符
Returns:
'''
'''递归去除字符串中特定的字符'''
if not str_in:
return str_in
if not isinstance(str_in,str):
str_in=str(str_in)
if remove_type==0:
if str_in[0]!=space_char:
return str_in
return cls.recursion_remove_char(str_in=str_in[1:],space_char=space_char,remove_type=remove_type)
elif remove_type==1:
if str_in[-1] != space_char:
return str_in
return cls.recursion_remove_char(str_in=str_in[:-1], space_char=space_char, remove_type=remove_type)
else:
str_in=cls.recursion_remove_char(str_in=str_in,space_char=space_char,remove_type=0)
str_in=cls.recursion_remove_char(str_in=str_in,space_char=space_char,remove_type=1)
return str_in
def isevaluatable(self,str_in):
'''判断字符是否可被eval'''
try:
ast.literal_eval(str_in)
return True
except :
return False
def str_output_case(self, str_in):
if not isinstance(str_in,str):
str_in = str(str_in)
out_str_tuple = (str_in, str_in.lower(), str_in.upper(), str_in.capitalize())
return out_str_tuple
def standard_str(self, str_in,**kwargs):
# '''将带有time关键字的参数放到字符串末尾，并去除参数sign'''
if not str_in:
return str_in
time_key=kwargs['time_key'] if kwargs.get('time_key') else 'time'
sign_key=kwargs['sign_key'] if kwargs.get('sign_key') else 'sign'
spilt_key=kwargs['spilt_key'] if kwargs.get('spilt_key') else '\n'
space_one=kwargs['space_one'] if kwargs.get('space_one') else '='
str_in_split=str_in.split(spilt_key)
str_in_split=[i for i in str_in_split if i]
# '''过滤参数中带有time、sign（根据实际情况修改）得参数'''
str_in_split_time=[i for i in str_in_split if i.startswith(f'{time_key}{space_one}') ]
str_in_split_sign=[i for i in str_in_split if i.startswith(f'{sign_key}{space_one}') ]
[ str_in_split.remove(i) for i in str_in_split_time ]
[ str_in_split.remove(i) for i in str_in_split_sign ]
str_in_split.extend(str_in_split_time)
str_in_split.extend(str_in_split_sign)
str_out = spilt_key.join(str_in_split)
return str_out
def str_to_dict(self, str_in, space_one=':', space_two='\n',is_params_sort=False,**kwargs):
# '''将str转换为dict输出'''
if not str_in:
return {}
if is_params_sort:
str_in = self.standard_str(str_in,spilt_key=space_two,space_one=space_one,**kwargs)
str_in_split = [ i.strip() for i in str_in.split(space_two) if i.strip()]
str_in_dict_keys=[i[:i.find(space_one)].strip() for i in str_in_split ]
str_in_dict_values=[ eval(i[i.find(space_one)+1:].strip()) if self.isevaluatable(i[i.find(space_one)+1:].strip())
else i[i.find(space_one) + 1:].strip() for i in str_in_split ]
str_in_dict=dict(zip(str_in_dict_keys,str_in_dict_values))
return str_in_dict
def dict_to_str(self, dict_in, space_one='=', space_two='&'):
# 将dict转换为str
if isinstance(dict_in, dict) and dict_in:
str_out = ''
for k, v in dict_in.items():
str_out += '{}{}{}{}'.format(k, space_one, v, space_two)
return self.recursion_remove_char(str_in=str_out,space_char=space_two,remove_type=1)
return ''
# 对字典进行排序
def sorted_dict(self,dict_in):
if isinstance(dict_in, dict):
__dict = dict(sorted(dict_in.items(), key=lambda x: x[0])) # 对字典进行排序
return __dict
return None
def request_data_to_str_postman(self, str_in, space_one=':', space_two='\n'):
# ''' 一个用于指定输出的postman的方法，去除空格'''
str_dict = self.str_to_dict(str_in)
str_out = self.dict_to_str(str_dict, space_one=space_one, space_two=space_two)
return str_out
# @exec_time_wrapper(round_num=10,module_obj=__file__,class_obj=sys._getframe().f_code.co_name,is_send_email=False)
def encry_main(self, str_in,encry_type='md5',upper=False,encoding='utf-8'):
# '''生成md5加密字符串'''
try:
str_out = eval(f"hashlib.{encry_type}()") # 采用md5加密
str_out.update(str(str_in).encode(encoding=encoding))
if upper :
return str_out.hexdigest().upper()
return str_out.hexdigest()
except:
raise Exception(f"不支持的加密方式：,目前支持的加密方式:{hashlib.algorithms_guaranteed}")
def url_join(self,base, url):
# 针对url进行拼接
url_join = urljoin(base, url)
url_array = urlparse(url_join)
path = normpath(url_array[2])
return urlunparse((url_array.scheme, url_array.netloc, path, url_array.params, url_array.query, url_array.fragment))
def generate_file_lineStr(self,join_str='\t',*args):
# 生成文件行字符串，传入数组，分割符，完成字符串组装
generate_str=''
for i in args:
generate_str+=str(i)+str(join_str)
return generate_str.strip()
def assert_to_pyobject(self, str_in):
# 将断言中的true/false/null，转换为python对象
str_dict = self.str_to_dict(str_in)
src_value_list=['true','false','null']
target_values_list=[True,False,None]
for str_dict_key in str_dict:
for index,src_value in enumerate(src_value_list):
if str_dict[str_dict_key]==src_value:
str_dict[str_dict_key]=target_values_list[index]
return str_dict
def pyobject_to_json_str(self,str_in,src_value_list=[],target_values_list=[]):
'''将py对象转换为json_str'''
src_value_list = ['True', 'False', 'None','"'] if not src_value_list else src_value_list
target_values_list = ['true', 'false', 'null',"'"] if not target_values_list else target_values_list
for index, src_value in enumerate(src_value_list):
str_in = str_in.replace(src_value, target_values_list[index])
return str_in
def json_str_to_pyobject(self,str_in):
# 将json_str中的true/false/null，转换为python对象
src_value_list=[' ','true','false','null','<NULL>']
target_values_list=['','True','False','None','None']
str_in=self.pyobject_to_json_str(str_in=str_in,src_value_list=src_value_list,target_values_list=target_values_list)
# 处理多个引号的字符
json_pyobj=eval(str_in) if self.isevaluatable(str_in) else str_in
if isinstance(json_pyobj,dict):
for i in json_pyobj:
json_pyobj[i]=eval(json_pyobj[i]) if self.isevaluatable(json_pyobj[i]) else json_pyobj[i]
return json_pyobj
def check_filename(self,file_name,ignore_chars:(list,tuple,set)=[],priority_matching_chars:(list,tuple,set)=[]):
'''
校验文件名称的方法，在 windows 中文件名不能包含('\','/','*','?','<','>','|') 字符
Args:
ignore_chars: 不需要被替换的字符集合，范围中chars
priority_matching_chars: 优先被匹配的字符集合，如果不为空，直接使用此集合定义的字符即进行替换
Returns:
'''
if priority_matching_chars:
for i in priority_matching_chars:
file_name=file_name.replace(i,'')
return file_name
chars=['\\','\/','*','?','<','>','|','\n','\b','\f','\t','\r','-',' ','.',':','[',']']
chars=[i for i in chars if i not in ignore_chars]
for i in chars:
file_name=file_name.replace(i,'')
return file_name
def get_project_rootpath(self):
"""
获取项目根目录。此函数的能力体现在，不论当前module被import到任何位置，都可以正确获取项目根目录
:return:
"""
path = os.path.realpath(os.curdir)
if sys.platform=='win32':
root_path=path[:path.find(':\\')+2]
else:
root_path=os.path.altsep
while path!=root_path:
# PyCharm项目中，'.idea'是必然存在的，且名称唯一
match_paths=['.idea','utils','handle','config']
is_match=all([i in os.listdir(path) for i in match_paths])
if is_match:
return path
path = os.path.dirname(path)
def is_contains_chinese(self,strs):
'''检测字符是否包含中文'''
for _char in strs:
if '\u4e00' <= _char <= '\u9fa5':
return True
return False
def append_params_key_isobject_to_extra_cases(self,case_params_key_obj,params_keys,extra_cases,
filter_func,cases_obj=[],is_append_extra_cases=True,params_key=""):
'''将params[key]->object的value添加到extra_cases'''
if not isinstance(case_params_key_obj, OrderedDict):
case_params_key_obj = OrderedDict(case_params_key_obj)
case_params_key_results = list(
AllPairs(case_params_key_obj, filter_func=filter_func))
for case_params_key in case_params_key_results:
case = self.dispose_allpairs_obj_str_to_dict(case_str=str(case_params_key),
params_keys=params_keys,
extra_cases=extra_cases,filter_func=filter_func)
if is_append_extra_cases:
cases_obj={params_key: case}
hash_case_obj=self.encry_main(str(cases_obj),encry_type='md5')
if not hash_case_obj in self.extra_cases_hash_lists:
self.extra_cases_hash_lists.append(hash_case_obj)
extra_cases.append(cases_obj)
else:
cases_obj.append(case)
# @exec_time_wrapper(round_num=10,module_obj=__file__,class_obj=sys._getframe().f_code.co_name,is_send_email=False)
def dispose_allpairs_obj_str_to_dict(self,case_str: str, params_keys,extra_cases,filter_func):
'''AllPairs对象转换为str，并最终处理为dict输出：并过滤掉参数key为object的key，存入extra_cases对象'''
def match_params_key_not_object(match_expression,case_str,space_comma):
'''
Args:
match_expression: 正则表达式
case_str: 需要被匹配的请求参数字符
is_match_object: 定义类型区分是否走object匹配
Returns:
'''
case_params_key_re = re.search(match_expression, case_str)
if not case_params_key_re:
return None
case_params_key_result = case_params_key_re.group()
return self.recursion_remove_char(str_in=case_params_key_result,space_char=space_comma,remove_type=1)
def match_params_key_object(case_params_key_result,params_key,space_one,space_comma,extra_cases,filter_func):
'''处理dict[key]->object专用'''
if not case_params_key_result.startswith("{}{}".format(params_key, space_one) + "{"):
return case_params_key_result
case_params_key_result_dict_str=case_params_key_result[len(params_key)+len(space_one):]
if case_params_key_result.endswith("}"+space_comma):
case_params_key_result_dict_str = case_params_key_result_dict_str[:-1]
case_params_key_result_dict = eval(case_params_key_result_dict_str) if \
self.isevaluatable(case_params_key_result_dict_str) else case_params_key_result_dict_str
'''将params[key]->object的value添加到extra_cases'''
self.append_params_key_isobject_to_extra_cases(case_params_key_obj=case_params_key_result_dict,
params_keys=list(case_params_key_result_dict.keys()),extra_cases=extra_cases,filter_func=filter_func,
cases_obj=[],is_append_extra_cases=True,params_key=params_key)
def match_params_key_object_main(match_expression,case_str,params_key,index,space_one,space_comma,extra_cases,filter_func,is_match_object=True):
'''完成处理dict[key]->object的主程序'''
if index < len(params_keys) - 1:
match_expression = f"{params_key}{space_one}.+,\s+{params_keys[index + 1]}"
case_params_key_result=match_params_key_not_object(match_expression=match_expression,case_str=case_str,space_comma=space_comma)
if index < len(params_keys) - 1:
case_params_key_result = case_params_key_result[:case_params_key_result.rfind(space_comma) + 1]
case_params_key_result=self.recursion_remove_char(str_in=case_params_key_result,space_char=space_comma,remove_type=1)
# 将key对应的值为object类型的key与值添加到cls.extra_cases
if is_match_object:
return match_params_key_object(case_params_key_result=case_params_key_result,params_key=params_key,
space_one=space_one,space_comma=space_comma,extra_cases=extra_cases,filter_func=filter_func)
return case_params_key_result
def merge_params_dict(params_keys, case_str,extra_cases,filter_func, space_one='=', space_two='\n',space_comma=','):
'''将将参数中所有的key与value转换为dict并合并为一个dict，
key对应的value为object时进行特殊处理（目前只支持key>json>key>not_json,即不支持三层及以上的json嵌套）'''
all_params_dict = {}
for index, params_key in enumerate(params_keys):
'''常规模式匹配字符，即key为不是最后一个且key对应的值不是object'''
match_expression_rule=f"{params_key}{space_one}\S+{space_comma}"
case_params_key_result=match_params_key_not_object(match_expression=match_expression_rule,case_str=case_str,space_comma=space_comma)
if case_params_key_result:
if case_params_key_result.startswith(f'{params_key}{space_one}[') : #and case_params_key_result.endswith(']')
'''特殊处理：key对应的值为list且元素为string的情况'''
match_expression_list_or_str=f"{params_key}{space_one}\S+{space_comma}"
case_params_key_result = match_params_key_object_main(
match_expression=match_expression_list_or_str, case_str=case_str,
params_key=params_key, index=index, space_one=space_one,
space_comma=space_comma, extra_cases=extra_cases, filter_func=filter_func, is_match_object=False)
else:
'''处理key对应的值为object的情况'''
match_expression_object = f"{params_key}{space_one}.+"
case_params_key_result=match_params_key_object_main(
match_expression=match_expression_object,case_str=case_str,
params_key=params_key,index=index,space_one=space_one,
space_comma=space_comma,extra_cases=extra_cases,filter_func=filter_func,is_match_object=True)
params_key_dict = self.str_to_dict(str_in=case_params_key_result, space_one=space_one,
space_two=space_two)
all_params_dict.update(params_key_dict)
return all_params_dict
if case_str.startswith("Pairs(") and case_str.endswith(")"):
case_str = case_str[len("Pairs("):-1]
all_params_dict = merge_params_dict(params_keys=params_keys, case_str=case_str, space_one='=', space_two='\n',
extra_cases=extra_cases,filter_func=filter_func,space_comma=',')
return all_params_dict
if __name__=="__main__":
ot=otherUtil()
str1="breakageType=['linear', 'alligator', 'pothole', 'raveling', 'explicit', 'subsidence', 'rut', 'seal', 'patch', 'horizontalCrack', 'verticalCrack']"
print(ot.str_to_dict(str_in=str1,space_one='=',space_two='\n'))
str1='..1.2.3...'
print(ot.recursion_remove_char(str_in=str1, space_char='.', remove_type=0))
print(ot.recursion_remove_char(str_in=str1, space_char='.', remove_type=1))
print(ot.recursion_remove_char(str_in=str1, space_char='.', remove_type=2))
if __name__=="__main__":
ot=otherUtil()