需求
接口测试，包含有多种测试场景，检查response。只是一次性测试，就不弄啥测试框架了哈，简单看看。
设计表格内容
包含request 和response需要检查的code、result_code、message等
代码主要设计如下部分
下面直接上代码
读excel数据
class Readdata():
def __init__(self):
nowpath = os.path.dirname(os.path.realpath(__file__))
# datapath = os.path.join(nowpath,'..' )
self.xlspath=nowpath +'/testdata.xlsx'
# print(self.xlspath)
if not os.path.exists(self.xlspath):
raise FileNotFoundError('The test data file does not exist')
def get_data(self,sheetname='testdata'):
file = xlrd.open_workbook(self.xlspath, 3)
cls = []
clsdict_list=[]
# Get the sheet that opens Excel
sheet = file.sheet_by_name(sheetname)
nrows = sheet.nrows
# Loop according to the number of lines
for i in range(nrows):
# print(sheet.row_values(i))
# If the first column of row i of the Excel sheet is not equal to casename then we add that row to cls[]
if sheet.row_values(i)[0] != u'casename':
data_dict=dict(zip(sheet.row_values(0),sheet.row_values(i)))
clsdict_list.append(data_dict)
cls.append(sheet.row_values(i))
return clsdict_list
封装request
'''
Created by Mia
'''
import json
import requests
import random
import traceback
from requests_toolbelt import MultipartEncoder
from log import log
class Request:
def get_request(self,url,data,headers=None,timeout=None):
'''
Get
:param url:
:param data:
:param headers:
:return:
'''
try:
response = requests.get(url=url,params=data,headers=headers,timeout=timeout)
time_rspconsume= response.elapsed.total_seconds()
response_dicts={}
response_dicts['code'] = response.status_code
response_dicts['body'] = response.json()
response_dicts['time_rspconsume'] = time_rspconsume
log.info(url + ' request_data: ' + str(json.loads(data)) + ' response_content: ' + str(response_dicts['body']))
return response_dicts
except TimeoutError:
log.error("TIme out!")
return None
except Exception as e:
log.error(traceback.format_exc())
return None
def post_requst(self, url,data ,headers,timeout):
'''
Post
:param url:
:param data:
:param headers:
:return:
'''
try:
response = requests.post(url=url,data=data,headers=headers,timeout=timeout)
print(response.text)
if response.status_code != 200:
return response.text
else:
time_rspconsume= response.elapsed.total_seconds()
response_dicts={}
response_dicts['code'] = response.status_code
response_dicts['body'] = response.json()
response_dicts['time_rspconsume'] = time_rspconsume
log.info(url + ' request_data: ' + data + ' response_content: ' + str(response_dicts['body']))
return response_dicts
except TimeoutError:
log.error("TIme out!")
return None
except Exception as e:
log.error(traceback.format_exc())
return response.text
def post_request_multipart(self,url,data,headers=None,timeout=None):
"""
Submit a Post request in Multipart/form-data format
:param url:
:param data:
:param headers:
:param file_parm:
:param file:
:param type:
:return:
"""
try:
m= MultipartEncoder(fields=data,boundary='-------------'+ str(random.randint(1e28,1e29)))
response= requests(url=url,data=m,headers=headers,timeout=timeout)
time_rspconsume= response.elapsed.total_seconds()
response_dicts={}
response_dicts['code'] = response.status_code
response_dicts['body'] = response.json()
response_dicts['time_rspconsume'] = time_rspconsume
log.info(url + ' request_data: ' + str(json.loads(data)) + ' response_content: ' + str(response_dicts['body']))
return response_dicts
except TimeoutError:
log.error("TIme out!")
return None
except Exception as e:
log.error(traceback.format_exc())
return None
def put_request(self,url,data,headers,timeout):
"""
Put
:param url:
:param data:
:param headers:
:return:
"""
try:
response=requests.put(url=url,data=data,headers=headers,timeout=timeout)
time_rspconsume= response.elapsed.total_seconds()
response_dicts={}
response_dicts['code'] = response.status_code
response_dicts['body'] = response.json()
response_dicts['time_rspconsume'] = time_rspconsume
log.info(url + ' request_data: ' + str(json.loads(data)) + ' response_content: ' + str(response_dicts['body']))
return response_dicts
except TimeoutError:
log.error("TIme out!")
return None
except Exception as e:
log.error(traceback.format_exc())
return None
def requests_main(self, url, data, method ,headers=None,timeout=None):
if method=='post':
result = self.post_requst(url,data,headers,timeout)
return result
elif method== 'get':
result = self.get_request(url,data,headers,timeout)
return result
elif method=='put':
result =self.put_request(url,data,headers,timeout)
return result
else:
log.error('Currently the form form is not compatible or has errors',)
return False
req=Request()
封装assertion
'''
Created by Mia
'''
from log import log
import json
import requests
class Assertions():
def assert_code(self,code, except_code):
log.info('*****begin assert code*******')
try:
assert int(code) == int(except_code)
log.info('code assertion pass,code=%d, except_code=%d',code,except_code)
return True
except:
log.error("Code error ， code is %s ，except_code is %s" %(str(code),str(except_code)))
raise
def assert_body(self,body,result_code,except_result_code):
log.info('*****begin assert body*******')
if isinstance(body,dict):
try:
assert int(except_result_code)==int(body[result_code])
log.info('result_code assertion pass, result_code= %d, except_result_code= %d' %(int(body[result_code]),int(except_result_code)))
return True
except:
log.error(result_code + ' is diff , result_code is %d , except_result_code is %d ' %(int(body[result_code]),int(except_result_code)))
raise
else:
raise
def assert_contain_text(self,body,expected_msg):
log.info('*****begin assert contain text*******')
try:
text = json.dumps(body,ensure_ascii=False)
assert expected_msg in text
log.info('excepted_msg assertion pass, text=%s, excepted_msg=%s',text,expected_msg)
return True
except:
log.error('Response body Does not contain expected_msg, expected_msg is %s' %expected_msg)
raise
def assert_body_key(self, body, expected_msg):
try:
if isinstance(body, str):
body_dict=json.loads(body)
for key in body_dict:
body_dict[key]=''
if isinstance(expected_msg):
expected_msg_dict = json.loads(body)
for key in expected_msg_dict:
expected_msg[key]=''
assert body_dict== expected_msg_dict
return True
except:
log.error('body_dict is diff , body_dict is %s , expected_msg_dict is ' %(body_dict, expected_msg_dict))
raise
ass= Assertions()
封装Log
'''
Created by Mia
'''
import logging ,os
class Logger(object):
def __init__(self):
self.cur_path = os.path.dirname(os.path.realpath(__file__))
self.log_dir = os.path.abspath(os.path.join(self.cur_path,'Log'))
if os.path.exists(self.log_dir):
self.logfile_path=os.path.join(self.log_dir,"log.log")
else:
os.mkdir(self.log_dir)
self.logfile_path=os.path.abspath(os.path.join(self.log_dir,"log.log"))
def get_log(self):
logger=logging.getLogger('V1API')
self.fmt = '%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(name)s - %(message)s'
self.formatter=logging.Formatter(self.fmt)
self.handler= logging.FileHandler(self.logfile_path,encoding='utf-8')
self.handler.setFormatter(self.formatter)
logger.addHandler(self.handler)
logger.setLevel(logging.DEBUG)
return logger
log=Logger().get_log()
最后再调用request 和 assertion，完工
def all_api(self):
data = self.get_data()
for i in data:
self.casename=i.get('casename')
self.host=i.get('host')
self.path = i.get('path')
# Convert a string to a dict using ast.literal_eval
self.headers = ast.literal_eval(i.get('headers'))
self.url = self.host+self.path
self.data=i.get('data')
self.method=i.get('method')
self.code=i.get('code')
self.except_code=int(i.get('except_code'))
self.result_code=i.get('result_code')
self.except_result_code=int(i.get('except_result_code'))
self.except_msg=i.get('except_msg')
self.is_run=i.get('is_run')
if self.is_run < 1:
pass
else:
resp=req.requests_main(self.url,self.data,self.method,headers=self.headers)
ass.assert_code(resp['code'],self.except_code)
ass.assert_body(resp['body'],self.result_code,self.except_result_code)
ass.assert_contain_text(resp['body'],self.except_msg)
运行一下，看下log成了，不用一遍遍改postman啦，就直接把测试数据放到Excel中，也有log方便回顾测试结果
