common.py
17.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import os
import operator
import configparser
import requests
import json
from commons.Logging import Logger
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
log=Logger()
def get_global_config(file,section, key):
current_path=os.path.dirname(__file__)
src_path=os.path.dirname(current_path)
# global_setting_path=src_path+'/config/global_data.conf'
global_setting_path=src_path+'/config/'+file+".conf"
#验证文件是否存在
file_path = os.path.exists(global_setting_path)
if file_path:
#获取文件的数据
if isinstance(key,int):
log.error("key of section cannot be int Type :<%r> "%str(key))
raise Exception("key of section cannot be int Type :<%r> "%str(key))
else:
config = configparser.ConfigParser()
config.read(global_setting_path,encoding="utf-8")
return config.get(section, key)
else:
log.error("File Not Exist :<%r> "%str(global_setting_path))
raise Exception("File Not Exist :<%r> "%str(global_setting_path))
def get_api_config(api,section,key):
current_path=os.path.dirname(__file__)
src_path=os.path.dirname(current_path)
api_path=src_path+"/config/api/"
api_config_path=api_path+api+".conf"
#验证文件是否存在
file_path = os.path.exists(api_config_path)
if file_path:
#获取文件的数据
if isinstance(key,int):
log.error("key of section cannot be int Type :<%r> "%str(key))
raise Exception("key of section cannot be int Type :<%r> "%str(key))
else:
config = configparser.ConfigParser()
config.read(api_config_path,encoding="utf-8")
return config.get(section, key)
else:
log.error("File Not Exist :<%r> "%str(api_config_path))
raise Exception("File Not Exist :<%r> "%str(api_config_path))
def set_global_config(file,section, key,value):
current_path=os.path.dirname(__file__)
src_path=os.path.dirname(current_path)
# global_setting_path=src_path+'/config/global_data.conf'
global_setting_path=src_path+'/config/'+file+".conf"
#验证文件是否存在
file_path = os.path.exists(global_setting_path)
if file_path:
#获取文件的数据
if isinstance(key,int):
log.error("key of section cannot be int Type :<%r> "%str(key))
raise Exception("key of section cannot be int Type :<%r> "%str(key))
else:
config = configparser.ConfigParser()
config.read(global_setting_path,encoding="utf-8")
config.set(section, key, value)
config.write(open(global_setting_path, "w"))
return config.get(section, key)
else:
log.error("File Not Exist :<%r> "%str(global_setting_path))
raise Exception("File Not Exist :<%r> "%str(global_setting_path))
def set_priority(priority_list):
#用户自定义优先级方法，设定有需要运行的用例优先级
#方法内定义global全局变量可以真正通过这个方法设置一个全局的值，如果global放在外面，方法内的变量都是临时的，
#接上面：即使名称和外面的全局变量相同，方法内的变量和外面的全局变量完全是两码事，外面的全局变量不会受干扰
global _global_priority
if priority_list:
_global_priority=priority_list
return _global_priority
else:
_global_priority=[1,2,3]
return _global_priority
def set_mark(case_mark):
#用户自定义优先级方法，设定有需要运行的用例优先级
global _global_mark
if case_mark:
_global_mark=case_mark
return _global_mark
else:
_global_mark=False
return _global_mark
def read_priority_list():
#需要重新设定一个方法来读取配置的全局优先级变量,且必须在全局变量初始化后使用，不然会报错
#如果直接再py文件中加入类似p=set_priority_list()的代码，之前用户自定义的优先级配置将会失效
priority_list=[1,2,3]
try:
return _global_priority
except Exception as e:
return priority_list
def read_case_mark():
#需要重新设定一个方法来读取配置的全局优先级变量,且必须在全局变量初始化后使用，不然会报错
#如果直接再py文件中加入类似p=set_priority_list()的代码，之前用户自定义的优先级配置将会失效
casemark=False
try:
return _global_mark
except Exception as e:
return casemark
def mark(tag=None,p=None):
#此处的try主要用于，当用户需要直接运行单个用例文件时，没提前_global_priority和_global_mark初始化
#如果不加try处理，那么_global_priority和_global_mark没有初始化时，单独运行一个引用了mark方法的用例文件会报错
try:
#用户没有输入任何值
if _global_priority == [1,2,3] and _global_mark == False :
return True
#只指定用例标签
elif _global_priority == [1,2,3] and _global_mark != False:
if tag == _global_mark:
return True
else:
return False
#只指定用例优先级
elif _global_priority != [1,2,3] and _global_mark == False:
if p in _global_priority:
return True
else:
return False
#同时指定优先级和标签
elif _global_priority != [1,2,3] and _global_mark != False:
if p in _global_priority and tag == _global_mark:
return True
else:
return False
#若直接在用例文件中运行那么_global_mark和_global_priority都不会被定义，会导致异常
#这里的try用来当上述两个变量未定义时，默认跑当前文件的用例
except Exception as e:
return True
def get_session(name,password):
pass
return "header"
def get_token(name,password):
pass
return "token"
def get_request_Url(api):
return get_api_config(api,'RequestUrl', 'url')
def get_request_Json(api):
return get_api_config(api,'JsonStructrue', 'requestJson')
# def post_request(api_url, argu, headers):
# #传值前最好要进行一个类型判断，以免数据类型没有转换正确，或者重复转换，导致调用接口时失败
# test_headers=json.loads(headers)
# r = requests.post(api_url, data=argu, headers=test_headers)
# return r
def dict_diff(a1,b1):
#提取出dict与str或list的异常对比场景
if type(a1)!= type(b1):
print("实际响应json的元素{}的类型{}和期望响应json元素{}的类型{}不一致".format(a1,type(a1),b1,type(b1)))
return False
d1=list(set(a1.keys()).difference(set(b1.keys())))
d2=list(set(b1.keys()).difference(set(a1.keys())))
if len(d1)==0 and len(d2)==0:
return True
else:
print("实际json比期望json多了键值",d1)
print("实际json比期望json少了键值",d2)
return False
def list_diff(a1,b1):
#提取出list与str或dict的异常对比场景
if type(a1)!= type(b1):
print("实际响应json的元素{}的类型{}和期望响应json元素{}的类型{}不一致".format(a1,type(a1),b1,type(b1)))
return False
elif len(a1)!=len(b1):
print("实际响应json列表长度{}和期望响应json列表长度{}不一致".format(a1,b1))
return False
else:
return True
#无法判断[[1],[2]]和[[1],[1]]的区别,只能判断列表长度和字典键值是否一致
#a为实际对比的json数据,b为期望的json数据
def compare_json(a,b):
#判断检查数据是否为列表格式的数据
if isinstance(a,list):
if list_diff(a,b)==True:
#为列表时必须进行排序,否则可能出现列表对比的位置不一致
a.sort()
b.sort()
for i in range(len(a)):
#判断列表中的指定的元素是否为字典格式
if compare_json(a[i],b[i]) == False:
return False
else:
return False
#判断检查数据是否为字典格式的数据
elif isinstance(a,dict):
if dict_diff(a,b)==True:
#如果第一层键值相同时,遍历检查内层所有键值
for key in a.keys():
#内层键值为列表数据类型时
if compare_json(a[key],b[key]) == False:
return False
else:
return False
#期望数据非dict和list类型时,直接进行对比,可能会导致{"code":"200"}，{"code":"201"}出现不同的情况，所以注释下面的结果
# elif operator.eq(a, b)==False:
# print("实际数据的{}和期望数据的{}不一致".format(a,b))
# return False
#接口返回值是可控的情况可以用下面这个方法
def cmp_json(actual_data,expect_data):
log.info("======对比期望和实际响应Json格式======")
try:
if len(expect_data)!=0 and len(actual_data)!=0:
expect_data=json.loads(expect_data)
result=compare_json(actual_data,expect_data)
if result==None:
log.info("响应数据和期望数据的Json格式一致")
return True
else:
log.error("66响应数据和期望数据的Json格式不一致")
return False
elif len(expect_data)!=0 and len(actual_data)==0:
log.error("对比数据中，实际响应数据为空")
assert False
elif len(expect_data)==0 and len(actual_data)!=0:
log.error("对比数据中，期望响应数据为空，请检查配置文件")
assert False
elif len(expect_data)==0 and len(actual_data)==0:
log.error("对比数据都为空")
assert False
except Exception as e:
log.error("对比响应数据和期望数据的Json格式 ERROR == %s"%e)
return False
def get_to_post(data):
#需要传入get格式的请求,如：id=638&driver=好司机&licensePlate=车辆信息,返回字典类型数据
if isinstance(data, str):
test={i.split("=",1)[0]:i.split("=")[1] for i in data.split("&")}
return test
else:
raise Exception("transformation data must be str type")
def post_to_get(data):
if isinstance(data, dict):
#字典类型,格式如{'id':'668','driver':'好司机'}或者{"id":"668","driver":"好司机"}
test=json.dumps(data)
test=test.replace(" ","").replace('":"', '=').replace('","', "&").replace('"', '').replace("{", "").replace("}", "")
return test.encode('utf_8').decode("unicode_escape")
elif isinstance(data, str):
if '"' in data:
#字符串类型,格式如str({'id':'668','driver':'好司机'})
test=data.replace(" ","").replace('":"', '=').replace('","', "&").replace('"', '').replace("{", "").replace("}", "")
return test
elif "'" in data:
#json字符串类型,格式如str({"id":"668","driver":"好司机"})
test=data.replace(" ","").replace("':'", "=").replace("','", "&").replace("'", "").replace("{", "").replace("}", "")
return test
else:
raise Exception("pls check your transformation data")
def get_request(api,caseNum):
#传值前最好要进行一个类型判断，以免数据类型没有转换正确，或者重复转换，导致调用接口时失败
if 'http' in api.replace(' ',''):
log.info(u"======直接调用GET请求%s======"%api)
r = requests.post(api, headers=caseNum,verify=False)
else:
URL=get_api_config(api,'RequstURL', 'url')
#获取header
headers=get_api_config(api,'RequstHeader', 'header')
if 'none' == headers:
#header=str(获取有session和token的header)
pass
headers=json.loads(headers)
#获取URI
params=get_api_config(api,caseNum, 'URL_params')
params=params.replace("[","").replace("]","")
URI=URL+params
#进行get请求
r = requests.get(URI,headers=headers)
if r.status_code == 200:
# 关闭请求 释放内存
r.close()
return r
else:
r.close()
log.error(u"Request code==%s ! Response Code is not 200,Pls Check Your Request Datas"%(r.status_code))
raise Exception(u"Request Fail! Response Code is not 200,Pls Check Your Request Datas")
return r
def post_request(api,caseNum,requestBody):
#设置重连次数
requests.adapters.DEFAULT_RETRIES = 5
# 设置连接活跃状态为False
s = requests.session()
s.keep_alive = False
#判断post方法使用方式
if 'http' in api.replace(' ',''):
log.info(u"======直接调用POST请求%s======"%api)
r = requests.post(api, headers=caseNum,data=requestBody,verify=False)
else:
log.info(u"======配置相关参数，调用POST请求%s======"%caseNum)
#传值前最好要进行一个类型判断，以免数据类型没有转换正确，或者重复转换，导致调用接口时失败
#获取请求地址数据
url=get_api_config(api,'RequstURL', 'url')
#获取请求header数据
header=get_api_config(api,'RequstHeader', 'header')
if 'none' == header:
#header=str(获取有session和token的header)
pass
#获取请求body数据
body=get_api_config(api,caseNum,requestBody)
#获取请求备注内容
notes=get_api_config(api,caseNum,'notes')
#获取请求期望json格式
expect_Response=get_api_config(api,caseNum,'expect_Response')
#把转换header类型
test_headers=json.loads(header)
#避免日志中的中文乱码，进行编码转换
test_body=json.dumps(json.loads(body)).decode('unicode_escape')
#调用post请求
r = requests.post(url, data=body, headers=test_headers,verify=False)
#进行日志记录
log.info(u"NUM:%s | 请求备注 ==%s"%(caseNum,notes))
log.info(u"NUM:%s | 请求URL==%s"%(caseNum,url))
log.info(u"NUM:%s | 请求body==%s"%(caseNum,test_body))
log.info(u"NUM:%s | 请求实际响应格式 ==%s"%(caseNum,r.content))
log.info(u"NUM:%s | 请求期望响应格式 ==%s"%(caseNum,expect_Response))
#判断请求是否成功
if r.status_code == 200:
# 关闭请求 释放内存
r.close()
return r
else:
r.close()
log.error(u"Request code==%s ! Response Code is not 200,Pls Check Your Request Datas"%(r.status_code))
raise Exception(u"Request Fail! Response Code is not 200,Pls Check Your Request Datas")
# a='https://gateway.fresh.nong12.com/getCityByPosition?lat=30.661286&lng=104.067624'
# b={
# "Host": "gateway.fresh.nong12.com",
# "Content-Type": "application/json",
# "referer": "https://servicewechat.com/wxd1405e5c40ff05db/0/page-frame.html",
# "apiusertoken": "dili-fresh-token",
# "charset": "utf-8",
# "Accept-Encoding": "gzip",
# "User-Agent": "Mozilla/5.0 (Linux; Android 7.0; HUAWEI NXT-AL10 Build/HUAWEINXT-AL10; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/59.0.3071.125 Mobile Safari/537.36 MicroMessenger/6.7.3.1360(0x2607033D) NetType/WIFI Language/zh_CN Process/appbrand2"
# }
# print get_request('getCityByPosition','getCityByPosition01').json()
# a= get_api_config('submitOrder', 'RequstHeader', 'header')
# b= get_global_config('basic_data_prepare', 'instert_customer')
# print(a)
# print(b)
# print type(a)
# a=json.loads(a)
# print a
# a=json.dumps(a)
# print a
# print a
# a=post_request('nearBy', 'nearBy01', 'requestBody')
# # b=get_api_config('listCoupon', 'listCoupon01', 'expect_Response')
# print a.json()
# print json_compare(b, a.json())
# a=['code', 'data', 'result', 'success']
# b=['code', 'data', 'result', 'success']
# print str(b)
# print json_compare(str(b), a)
# print a.content
# a=post_request('www.baidu.com/',"" ,"" )
# print a.headers
# print type(a)
# b=json.loads(a)
# c=json.dumps(json.loads(a)).decode('unicode_escape')
# print c
# a={"code":"200","data":{"code":"200","data":[],"result":"OK","success":1},"result":"OK","success":2}
# b={"code":"200","data":{"code":"200","data":[{"couponBizType":1,"couponCode":"004","couponConditionValue":0,"couponDescription":"auto","couponDiscount":1000,"couponDuration":10,"couponName":"成勇测试01","couponTimeEnd":"2025-01-26 23:59:59","couponTimeSt":"2019-01-17 00:00:00","couponType":2,"created":"2019-01-17 15:14:15","customerCode":"00010966","id":10000,"isUsable":2,"state":1,"version":1}],"result":"OK","success":2},"result":"OK","success":2}
# print type(a),type(b)
# prjson_cmp(a, b)