gfwleak/dongxiaoyan-tsg-autotest
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

上传report纯接口条件全覆盖方法

Browse Source
This commit is contained in:
byb11
2021-04-25 13:46:54 +08:00
parent bf4216dbc9
commit f4ea59dd71
1 changed files with 330 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

499
04-CustomLibrary/Custometest/ReportSchema.py
View File

@@ -41,22 +41,17 @@ def schema(schemauerl,token,logtype):
#获取json串中groupColumnList的值 #获取json串中groupColumnList的值
def groupby(schemajson,logtype,testpoint): def groupby(schemajson,logtype,testpoint):
#生成随机数
# securitynb=random.randint(1,33)
# proxynb=random.randint(1,23)
# sessionnb=random.randint(1,24)
# Radiusnb=random.randint(1,4)
#取出group值得列表
dimensions=schemajson["data"]["doc"]["schema_query"]["dimensions"] dimensions=schemajson["data"]["doc"]["schema_query"]["dimensions"]
dimensions.append("common_recv_time"); dimensions.append("common_recv_time");
randomstr_1=[] randomstr_1=[]
if testpoint == "GroupBy": if testpoint == "DataBindings":
randomstr_1=dimensions randomstr_1.append("common_recv_time")
if logtype == "security_event_log" or logtype == "connection_record_log" or logtype == "voip_record_log" :
randomstr_1.remove("common_start_time")
randomstr_1.remove("common_end_time")
else: else:
randomstr_1= random.sample(dimensions, 4) randomstr_1=random.sample(dimensions, 4)
if logtype == "security_event_log" or logtype == "connection_record_log" or logtype == "voip_record_log" :
randomstr_1.remove("common_start_time")
randomstr_1.remove("common_end_time")
#定义grp为返回值group的列表 #定义grp为返回值group的列表
grp=[] grp=[]
for i in randomstr_1: for i in randomstr_1:
@@ -68,16 +63,16 @@ def groupby(schemajson,logtype,testpoint):
return re return re
#获取json串中queryColumnList的值 #获取json串中queryColumnList的值
def DataBindings(schemajson,randomstr_1,testpoint): def DataBindings(schemajson,randomstr_1,testpoint,field):
#生成queryColumnList列表 #生成queryColumnList列表
metrics=schemajson["data"]["doc"]["schema_query"]["metrics"] metrics=schemajson["data"]["doc"]["schema_query"]["metrics"]
metrics.append("common_log_id") metrics.append("common_log_id")
#在列表里随机去除元素 #在列表里随机元素
randomstr_2=[] randomstr_2=[]
if testpoint == "DataBindings" : if testpoint == "DataBindings":
randomstr_2=metrics randomstr_2.append(field)
else : else:
randomstr_2= random.sample(metrics, 6) randomstr_2=random.sample(metrics,6)
#在聚合列表中去掉groupby中的重复的元素 #在聚合列表中去掉groupby中的重复的元素
randomstr_3=array_diff(randomstr_2,randomstr_1) randomstr_3=array_diff(randomstr_2,randomstr_1)
#将groupby中元素添加到串中 #将groupby中元素添加到串中
@@ -88,20 +83,40 @@ def DataBindings(schemajson,randomstr_1,testpoint):
fields = schemajson["data"]["fields"] fields = schemajson["data"]["fields"]
list_1=["sum","min","max","avg","count"] list_1=["sum","min","max","avg","count"]
list_2=["count","count_distinct"] list_2=["count","count_distinct"]
for i in randomstr_3: if testpoint == "DataBindings":
for j in fields: for i in randomstr_3:
if i == j["name"] : for j in fields:
jtype=j["type"] if i == j["name"] :
if jtype == "int" or jtype == "long" or jtype == "float" or jtype == "double": jtype=j["type"]
radomlist=random.sample(list_1,1) label=i
randomstr_4={"name":i,"expression":radomlist[0]} sun=1
qul.append(randomstr_4) if jtype == "int" or jtype == "long" or jtype == "float" or jtype == "double":
elif jtype == "randomstring" or jtype == "date" or jtype == "timestamp" or jtype == "string": for Aggregate in list_1:
randomlist=random.sample(list_2,1) randomstr_4={"name":i,"expression":Aggregate,"label":label}
randomstr_4={"name":i,"expression":randomlist[0]} qul.append(randomstr_4)
qul.append(randomstr_4) label=label+str(sun)
sun+=1
elif jtype == "randomstring" or jtype == "date" or jtype == "timestamp" or jtype == "string":
for Aggregate in list_2:
randomstr_4={"name":i,"expression":Aggregate,"label":label}
qul.append(randomstr_4)
label = label + str(sun)
sun += 1
else:
for i in randomstr_3:
for j in fields:
if i == j["name"]:
jtype = j["type"]
if jtype == "int" or jtype == "long" or jtype == "float" or jtype == "double":
radomlist = random.sample(list_1, 1)
randomstr_4 = {"name": i, "expression": radomlist[0]}
qul.append(randomstr_4)
elif jtype == "randomstring" or jtype == "date" or jtype == "timestamp" or jtype == "string":
randomlist = random.sample(list_2, 1)
randomstr_4 = {"name": i, "expression": randomlist[0]}
qul.append(randomstr_4)
print("DataBindings",qul) print("DataBindings",qul)
return qul return qul
@@ -120,15 +135,15 @@ def array_diff(a, b):
#返回序列c，c就是列表a去除列表b之后的元素 #返回序列c，c就是列表a去除列表b之后的元素
return c return c
def filterCondition(schemajson,testpoint): def filterCondition(schemajson,testpoint,field):
number = random.randint(0,100000) number = random.randint(0,100000)
randomstr= random.choice('abcdefghijklmnopqrstuvwxyz') randomstr= random.choice('abcdefghijklmnopqrstuvwxyz')
schemafilters=schemajson["data"]["doc"]["schema_query"]["filters"] schemafilters=schemajson["data"]["doc"]["schema_query"]["filters"]
list1=[] list1=[]
if testpoint== "Filter": if testpoint=="Filter":
list1=schemafilters list1.append(field)
else: else:
list1= random.sample(schemafilters,3) list1=random.sample(schemafilters, 4)
#获取不同属性支持的部不同操作 #获取不同属性支持的部不同操作
fields = schemajson["data"]["fields"] fields = schemajson["data"]["fields"]
operator = schemajson["data"]["doc"]["schema_query"]["references"]["operator"] operator = schemajson["data"]["doc"]["schema_query"]["references"]["operator"]
@@ -146,27 +161,45 @@ def filterCondition(schemajson,testpoint):
if type1 == "int" or type1 == "long": if type1 == "int" or type1 == "long":
orConditions_list=[] orConditions_list=[]
Operator=["=","!=",">","<",">=","<="] Operator=["=","!=",">","<",">=","<="]
randomOperator= random.sample(Operator, 1) if testpoint=="Filter":
value=[str(number)] for op in Operator:
Field={"name":name,"expression":randomOperator[0],"value":value,"type":type1} value=[str(number)]
orConditions_list.append(Field) Field={"name":name,"expression":op,"value":value,"type":type1}
orConditions_list.append(Field)
else:
randomOperator= random.sample(Operator, 1)
value=[str(number)]
Field={"name":name,"expression":randomOperator[0],"value":value,"type":type1}
orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list} orConditions={"orConditions":orConditions_list}
andConditions.append(orConditions) andConditions.append(orConditions)
elif type1 == "string": elif type1 == "string":
orConditions_list=[] orConditions_list=[]
Operator=["=","!=","Like","Not Like","notEmpty"] Operator=["=","!=","Like","Not Like","notEmpty","empty"]
randomOperator_1= random.sample(Operator, 1) if testpoint=="Filter":
randomOperator=randomOperator_1[0] for op in Operator:
value=[] value=[]
if randomOperator == "=" or randomOperator == "!=": if op == "=" or op == "!=":
value.append(str(number)) value.append(str(number))
elif randomOperator == "Like" or randomOperator == "Not Like": elif op == "Like" or op == "Not Like":
value.append(randomstr) value.append(randomstr)
elif randomOperator=="notEmpty": elif op=="notEmpty" or op == "empty":
value=[] value=[]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1} Field={"name":name,"expression":op,"value":value,"type":type1}
orConditions_list.append(Field) orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list} else:
randomOperator_1 = random.sample(Operator, 1)
randomOperator = randomOperator_1[0]
value = []
if randomOperator == "=" or randomOperator == "!=":
value.append(str(number))
elif randomOperator == "Like" or randomOperator == "Not Like":
value.append(randomstr)
elif randomOperator == "notEmpty":
value = []
Field = {"name": name, "expression": randomOperator, "value": value, "type": type1}
orConditions_list.append(Field)
orConditions = {"orConditions": orConditions_list}
andConditions.append(orConditions) andConditions.append(orConditions)
else: else:
@@ -175,26 +208,45 @@ def filterCondition(schemajson,testpoint):
if type1 == "int" or type1 == "long": if type1 == "int" or type1 == "long":
orConditions_list=[] orConditions_list=[]
Operator=["=","!=",">","<",">=","<="] Operator=["=","!=",">","<",">=","<="]
randomOperator= random.sample(Operator, 1) if testpoint == "Filter":
value=[str(number)] for op in Operator:
Field={"name":name,"expression":randomOperator[0],"value":value,"type":type1} value=[str(number)]
orConditions_list.append(Field) Field={"name":name,"expression":op,"value":value,"type":type1}
orConditions_list.append(Field)
else:
randomOperator= random.sample(Operator, 1)
value=[str(number)]
Field={"name":name,"expression":randomOperator[0],"value":value,"type":type1}
orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list} orConditions={"orConditions":orConditions_list}
andConditions.append(orConditions) andConditions.append(orConditions)
elif type1 == "string": elif type1 == "string":
orConditions_list=[] orConditions_list=[]
Operator=["=","!=","Like","Not Like","notEmpty"] Operator=["=","!=","Like","Not Like","notEmpty","empty"]
randomOperator_1= random.sample(Operator, 1) if testpoint == "Filter":
randomOperator=randomOperator_1[0] for op in Operator:
value=[] randomOperator = op
if randomOperator == "=" or randomOperator == "!=": value = []
value.append(str(number)) if randomOperator == "=" or randomOperator == "!=":
elif randomOperator == "Like" or randomOperator == "Not Like": value.append(str(number))
value.append(randomstr) elif randomOperator == "Like" or randomOperator == "Not Like":
elif randomOperator=="notEmpty": value.append(randomstr)
elif randomOperator == "notEmpty":
value = []
Field = {"name": name, "expression": randomOperator, "value": value, "type": type1}
orConditions_list.append(Field)
else:
randomOperator_1= random.sample(Operator, 1)
randomOperator=randomOperator_1[0]
value=[] value=[]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1} if randomOperator == "=" or randomOperator == "!=":
orConditions_list.append(Field) value.append(str(number))
elif randomOperator == "Like" or randomOperator == "Not Like":
value.append(randomstr)
elif randomOperator=="notEmpty":
value=[]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1}
orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list} orConditions={"orConditions":orConditions_list}
andConditions.append(orConditions) andConditions.append(orConditions)
@@ -203,70 +255,123 @@ def filterCondition(schemajson,testpoint):
conrandomstraints=k["doc"]["constraints"] conrandomstraints=k["doc"]["constraints"]
type1 = k["type"] type1 = k["type"]
if type1 == "int" or type1 == "long": if type1 == "int" or type1 == "long":
orConditions_list = []
Operator=["=","!=",">","<",">=","<="] Operator=["=","!=",">","<",">=","<="]
randomOperator_1= random.sample(Operator, 1) if testpoint == "Filter":
randomOperator=randomOperator_1[0] for op in Operator:
if conrandomstraints["type"] == "timestamp": randomOperator = op
#获取当前时间戳 if conrandomstraints["type"] == "timestamp":
t = int(time.time()) # 获取当前时间戳
orConditions_list=[] t = int(time.time())
value=[str(t)] value = [str(t)]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1} Field = {"name": name, "expression": randomOperator, "value": value,
orConditions_list.append(Field) "type": type1}
orConditions={"orConditions":orConditions_list} orConditions_list.append(Field)
andConditions.append(orConditions) else:
elif type1 == "string":
Operator=["=","!=","Like","Not Like","notEmpty"]
randomOperator_1= random.sample(Operator, 1)
randomOperator=randomOperator_1[0]
if conrandomstraints["type"] == "ip":
orConditions_list=[]
#获取ip
ip =random_ipv4()
value=[]
if randomOperator == "=" or randomOperator == "!=":
value.append(ip)
elif randomOperator == "Like" or randomOperator == "Not Like":
value.append(ip)
elif randomOperator=="notEmpty":
value=[]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1}
orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list}
andConditions.append(orConditions)
elif conrandomstraints["type"] == "email":
orConditions_list=[]
Operator=["=","!=","Like","Not Like","notEmpty"]
randomOperator_1= random.sample(Operator, 1) randomOperator_1= random.sample(Operator, 1)
randomOperator=randomOperator_1[0] randomOperator=randomOperator_1[0]
#获取ip if conrandomstraints["type"] == "timestamp":
emil =RandomEmail() #获取当前时间戳
value=[] t = int(time.time())
if randomOperator == "=" or randomOperator == "!=": value=[str(t)]
value.append(emil) Field={"name":name,"expression":randomOperator,"value":value,"type":type1}
elif randomOperator == "Like" or randomOperator == "Not Like": orConditions_list.append(Field)
value.append(emil) orConditions={"orConditions":orConditions_list}
elif randomOperator=="notEmpty": andConditions.append(orConditions)
elif type1 == "string":
orConditions_list = []
Operator=["=","!=","Like","Not Like","notEmpty","empty"]
if testpoint == "Filter":
if conrandomstraints["type"] == "ip":
for op in Operator:
# 获取ip
ip = random_ipv4()
value = []
if op == "=" or op == "!=":
value.append(ip)
elif op == "Like" or op == "Not Like":
value.append(ip)
elif op == "notEmpty":
value = []
Field = {"name": name, "expression": op, "value": value,"type": type1}
orConditions_list.append(Field)
elif conrandomstraints["type"] == "email":
for op in Operator:
randomOperator = op
Operator = ["=", "!=", "Like", "Not Like", "notEmpty","empty"]
randomOperator_1 = random.sample(Operator, 1)
randomOperator = randomOperator_1[0]
# 获取ip
emil = RandomEmail()
value = []
if randomOperator == "=" or randomOperator == "!=":
value.append(emil)
elif randomOperator == "Like" or randomOperator == "Not Like":
value.append(emil)
elif randomOperator == "notEmpty":
value = []
Field = {"name": name, "expression": randomOperator, "value": value,
"type": type1}
orConditions_list.append(Field)
else:
randomOperator_1= random.sample(Operator, 1)
randomOperator=randomOperator_1[0]
if conrandomstraints["type"] == "ip":
#获取ip
ip =random_ipv4()
value=[] value=[]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1} if randomOperator == "=" or randomOperator == "!=":
orConditions_list.append(Field) value.append(ip)
orConditions={"orConditions":orConditions_list} elif randomOperator == "Like" or randomOperator == "Not Like":
andConditions.append(orConditions) value.append(ip)
elif randomOperator=="notEmpty":
value=[]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1}
orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list}
andConditions.append(orConditions)
elif conrandomstraints["type"] == "email":
Operator=["=","!=","Like","Not Like","notEmpty","empty"]
randomOperator_1= random.sample(Operator, 1)
randomOperator=randomOperator_1[0]
#获取ip
emil =RandomEmail()
value=[]
if randomOperator == "=" or randomOperator == "!=":
value.append(emil)
elif randomOperator == "Like" or randomOperator == "Not Like":
value.append(emil)
elif randomOperator=="notEmpty":
value=[]
Field={"name":name,"expression":randomOperator,"value":value,"type":type1}
orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list}
andConditions.append(orConditions)
else: else:
type1 = k["type"] type1 = k["type"]
orConditions_list=[] orConditions_list=[]
operator1 = k["doc"]["constraints"]["operator_functions"] operator1 = k["doc"]["constraints"]["operator_functions"]
operator2 = operator1.split(",") operator2 = operator1.split(",")
operator3=random.sample(operator2,1) if testpoint == "Filter":
operatordata=k["doc"]["data"] for op in operator2:
code=[] operatordata = k["doc"]["data"]
for i in operatordata: code = []
code_1=i["code"] for i in operatordata:
code.append(code_1) code_1 = i["code"]
code2=random.sample(code, 1) code.append(code_1)
Field={"name":name,"expression":operator3[0],"value":code2,"type":type1} for co in code:
orConditions_list.append(Field) Field = {"name": name, "expression": op, "value": co, "type": type1}
orConditions_list.append(Field)
else:
operator3=random.sample(operator2,1)
operatordata=k["doc"]["data"]
code=[]
for i in operatordata:
code_1=i["code"]
code.append(code_1)
code2=random.sample(code, 1)
Field={"name":name,"expression":operator3[0],"value":code2,"type":type1}
orConditions_list.append(Field)
orConditions={"orConditions":orConditions_list} orConditions={"orConditions":orConditions_list}
andConditions.append(orConditions) andConditions.append(orConditions)
filterCondition={"andConditions":andConditions} filterCondition={"andConditions":andConditions}
@@ -274,18 +379,19 @@ def filterCondition(schemajson,testpoint):
return filterCondition return filterCondition
#获取having条件的串 #获取having条件的串
def havingjson(schemajson,testpoint): def havingjson(schemajson,testpoint,field):
number = random.randint(0,100000) number = random.randint(0,100000)
schemametrics=schemajson["data"]["doc"]["schema_query"]["metrics"] schemametrics=schemajson["data"]["doc"]["schema_query"]["metrics"]
aggregation = schemajson["data"]["doc"]["schema_query"]["references"]["aggregation"] aggregation = schemajson["data"]["doc"]["schema_query"]["references"]["aggregation"]
schemametrics.append("common_log_id") schemametrics.append("common_log_id")
metricslist=[] metricslist=[]
if testpoint == "Having": if testpoint == "Having":
metricslist=schemametrics metricslist.append(field)
else: else:
metricslist= random.sample(schemametrics,3) metricslist=random.sample(schemametrics, 4)
fields = schemajson["data"]["fields"] fields = schemajson["data"]["fields"]
operator=["=","!=",">","<",">=","<="] operator=["=","!=",">","<",">=","<="]
Aggregate=["COUNT","AVG","SUM","MAX","MIN"]
andConditions_list=[] andConditions_list=[]
#遍历的到的having条件列表 #遍历的到的having条件列表
for i in metricslist: for i in metricslist:
@@ -296,29 +402,44 @@ def havingjson(schemajson,testpoint):
for v in aggregation: for v in aggregation:
if type1 == v["type"]: if type1 == v["type"]:
orConditions_list=[] orConditions_list=[]
functionsstr=v["functions"] if v["type"] != "string":
functionslist = functionsstr.split(",") functionslist=Aggregate
functions_1=random.sample(functionslist, 1) else:
if functions_1=="COUNT_DISTINCT" and type1 != "string": functionsstr=v["functions"]
functionslist = functionsstr.split(",")
if field == "common_log_id":
functionslist=["COUNT"]
if testpoint == "Having":
for functions_1 in functionslist:
for operator_1 in operator:
havingdict = {"name": name, "function": str.lower(functions_1),
"expression": operator_1, "value": str(number)}
orConditions_list.append(havingdict)
orConditions = {"orConditions": orConditions_list}
andConditions_list.append(orConditions)
else:
functions_1=random.sample(functionslist, 1) functions_1=random.sample(functionslist, 1)
operator_1=random.sample(operator, 1) if functions_1=="COUNT_DISTINCT" and type1 != "string":
havingdict={"name":name,"function":str.lower(functions_1[0]),"expression":operator_1[0],"value":str(number)} functions_1=random.sample(functionslist, 1)
orConditions_list.append(havingdict) operator_1=random.sample(operator, 1)
orConditions={"orConditions":orConditions_list}
andConditions_list.append(orConditions) havingdict={"name":name,"function":str.lower(functions_1[0]),"expression":operator_1[0],"value":str(number)}
orConditions_list.append(havingdict)
orConditions={"orConditions":orConditions_list}
andConditions_list.append(orConditions)
havingCondition={"andConditions":andConditions_list} havingCondition={"andConditions":andConditions_list}
print("having",havingCondition) print("having",havingCondition)
return havingCondition return havingCondition
#拼接字符串 #拼接字符串
def datasetjson(schemauerl,token,testname,logtype,testpoint): def datasetjson(schemauerl,token,testname,logtype,testpoint,field):
schema_new=schema(schemauerl,token,logtype) schema_new=schema(schemauerl,token,logtype)
group_re=groupby(schema_new,logtype,testpoint) group_re=groupby(schema_new,logtype,testpoint)
groupColumnList=group_re[0] groupColumnList=group_re[0]
group_randomstr=group_re[1] group_randomstr=group_re[1]
queryColumnList=DataBindings(schema_new,group_randomstr,testpoint) queryColumnList=DataBindings(schema_new,group_randomstr,testpoint,field)
filterCondition_1=filterCondition(schema_new,testpoint) filterCondition_1=filterCondition(schema_new,testpoint,field)
havingjson_1=havingjson(schema_new,testpoint) havingjson_1=havingjson(schema_new,testpoint,field)
datasetdict = { datasetdict = {
"list": { "list": {
"name":testname, "name":testname,
@@ -494,52 +615,92 @@ def Reportsjson(chartId,testname):
print("reportjson",json.dumps(reportJobdct_2)) print("reportjson",json.dumps(reportJobdct_2))
return json.dumps(reportJobdct_2) return json.dumps(reportJobdct_2)
def ReportInterfaceTest(schemaurl,token,dataseturl,charurl,repporturl,datasetgeturl,chargeturl,testname,logtype,testpoint): def ReportInterfaceTest(schemaurl,token,dataseturl,charurl,repporturl,datasetgeturl,chargeturl,testname,logtype,testpoint,field):
headers = {"Content-Type": "application/json","Authorization": token} headers = {"Content-Type": "application/json","Authorization": token}
#dataset生成json串并发送请求 #dataset生成json串并发送请求
_datasetjson=datasetjson(schemaurl,token,testname,logtype,testpoint) _datasetjson=datasetjson(schemaurl,token,testname,logtype,testpoint,field)
response1 = requests.post(url=dataseturl, data=_datasetjson, headers=headers) response1 = requests.post(url=dataseturl, data=_datasetjson, headers=headers)
print("返回数据1",response1) print("返回数据1",response1)
code = response1.json()["code"] code = response1.json()["code"]
print("datasetcode:",code) print("datasetcode:",code)
assert code == 200 assert code == 200
#获取dataset的id
# 获取dataset的id
datasetget=requests.get(url=datasetgeturl,headers=headers) datasetget=requests.get(url=datasetgeturl,headers=headers)
dasetget=datasetget.json() dasetget=datasetget.json()
datesetid=dasetget["data"]["list"][0]["id"] datesetid=dasetget["data"]["list"][0]["id"]
_datasetjson=json.loads(_datasetjson) # Deleteinterfaces(dataseturl,token,datesetid)
queryColumnList=_datasetjson["list"]["queryColumnList"] # _datasetjson=json.loads(_datasetjson)
groupColumnList=_datasetjson["list"]["groupColumnList"] # queryColumnList=_datasetjson["list"]["queryColumnList"]
# groupColumnList=_datasetjson["list"]["groupColumnList"]
#生成charlibrariesjson串 #生成charlibrariesjson串
charlibrariesjson=charjson(schemaurl, token,queryColumnList,groupColumnList,datesetid,testname,logtype) # charlibrariesjson=charjson(schemaurl, token,queryColumnList,groupColumnList,datesetid,testname,logtype)
response2 = requests.post(url=charurl, data=charlibrariesjson, headers=headers) # response2 = requests.post(url=charurl, data=charlibrariesjson, headers=headers)
code = response2.json()["code"] # code = response2.json()["code"]
assert code == 200 # assert code == 200
#
#获取char libraries的id # #获取char libraries的id
charget=requests.get(url=chargeturl,headers=headers) # charget=requests.get(url=chargeturl,headers=headers)
charget=charget.json() # charget=charget.json()
charid=charget["data"]["list"][0]["id"] # charid=charget["data"]["list"][0]["id"]
#
#report生成json串并发送请求 # #report生成json串并发送请求
reportjson=Reportsjson(charid,testname) # reportjson=Reportsjson(charid,testname)
response3 = requests.post(url=repporturl, data=reportjson, headers=headers) # response3 = requests.post(url=repporturl, data=reportjson, headers=headers)
code = response3.json()["code"] # code = response3.json()["code"]
assert code == 200 # assert code == 200
#
# #循环调用ReportInterfaceTest方法
# def ReportTest(host,token,dataseturl,charurl,repporturl,logtypelist):
# for logtype in logtypelist:
# testname="Report"+logtype
# datasetgeturl=dataseturl+"?pageSize=20&pageNo=1&id=&name="+testname+"&logType=&opStartTime=&opEndTime=&opUser="
# chargeturl=charurl+"?pageSize=20&pageNo=1&id=&name="+testname+"&opUser="
# schemaurl="http://"+host+":8080/v1/log/schema?logType="+logtype
# ReportInterfaceTest(schemaurl,token,dataseturl,charurl,repporturl,datasetgeturl,chargeturl,testname,logtype)
def Deleteinterfaces(url,token,id):
headers = {"Content-Type": "application/json","Authorization": token}
datedict={"ids":[id]}
datajson=json.dumps(datedict)
response1 = requests.delete(url=url, data=datajson, headers=headers)
#循环调用ReportInterfaceTest方法 def ReportPositiveTest(host,port,token,dataseturl,charurl,repporturl,logtypelist):
def ReportTest(host,token,dataseturl,charurl,repporturl,logtypelist,testpointlist): # testpoint=["DataBindings","Filter","Having"]
testpoint=["DataBindings"]
for logtype in logtypelist: for logtype in logtypelist:
for testpoint in testpointlist: schemaurl="http://"+host+":"+port+"/v1/log/schema?logType="+logtype
testname="Report"+logtype+testpoint schema_new=schema(schemaurl,token,logtype)
datasetgeturl=dataseturl+"?pageSize=20&pageNo=1&id=&name="+testname+"&logType=&opStartTime=&opEndTime=&opUser=" metrics = schema_new["data"]["doc"]["schema_query"]["metrics"]
chargeturl=charurl+"?pageSize=20&pageNo=1&id=&name="+testname+"&opUser=" schemafilters = schema_new["data"]["doc"]["schema_query"]["filters"]
schemaurl="http://"+host+":8080/v1/log/schema?logType="+logtype metrics.append("common_log_id")
ReportInterfaceTest(schemaurl,token,dataseturl,charurl,repporturl,datasetgeturl,chargeturl,testname,logtype,testpoint) for j in testpoint:
if j == "DataBindings":
for filter in metrics:
testname="Report"+logtype+j+filter
dataset_geturl=dataseturl+"?pageSize=20&pageNo=1&id=&name="+testname+"&logType=&opStartTime=&opEndTime=&opUser="
char_geturl=charurl+"?pageSize=20&pageNo=1&id=&name="+testname+"&opUser="
ReportInterfaceTest(schemaurl,token,dataseturl,charurl,repporturl,dataset_geturl,char_geturl,testname,logtype,j,filter)
if j == "Filter" :
for filter in schemafilters:
testname="Report"+logtype+j+filter
dataset_geturl=dataseturl+"?pageSize=20&pageNo=1&id=&name="+testname+"&logType=&opStartTime=&opEndTime=&opUser="
char_geturl=charurl+"?pageSize=20&pageNo=1&id=&name="+testname+"&opUser="
ReportInterfaceTest(schemaurl,token,dataseturl,charurl,repporturl,dataset_geturl,char_geturl,testname,logtype,j,filter)
if j == "Having" :
for filter in metrics:
testname="Report"+logtype+j+filter
dataset_geturl=dataseturl+"?pageSize=20&pageNo=1&id=&name="+testname+"&logType=&opStartTime=&opEndTime=&opUser="
char_geturl=charurl+"?pageSize=20&pageNo=1&id=&name="+testname+"&opUser="
ReportInterfaceTest(schemaurl,token,dataseturl,charurl,repporturl,dataset_geturl,char_geturl,testname,logtype,j,filter)