get_data.py

封装：

#coding:utf-8from util.operation_excel import OperationExcelimport data_configfrom util.operation_json import OperationJsonfrom data.dependdata import DependentDataimport jsonclass GetData:def __init__(self):self.opera_excel = OperationExcel()# 获取excel行数，就是case的个数def get_case_lines(self):return self.opera_excel.get_lines()#是否执行def get_is_run(self,row):flag = Nonecol = int(data_config.get_run())run_model = self.opera_excel.get_cell_value(row,col)if run_model == 'yes':flag = Trueelse:flag = Falsereturn flag# 是否携带headerdef is_header(self,row):col = int(data_config.get_header())header = self.opera_excel.get_cell_value(row, col)return headerdef header1(self,row):col = int(data_config.get_header())header = self.opera_excel.get_cell_value(row, col)header = data_config.get_header_value()return headerdef headervalue(self,row):col = int(data_config.get_header())header = self.opera_excel.get_cell_value(row,col)header = data_config.get_aut_value()return header# return header, data_config.get_header_value()# def is_header(self,row):#header = data_config.get_header_value()#return header#if header != '':# return header#else:# return None# 获取请求方式def get_request_method(self,row):col = int(data_config.get_request_way())request_method = self.opera_excel.get_cell_value(row,col)return request_method# 获取urldef get_request_url(self,row):col = int(data_config.get_url())url = self.opera_excel.get_cell_value(row,col)return url# 获取请求数据def get_request_data(self,row):col = int(data_config.get_data())data = self.opera_excel.get_cell_value(row,col)if data == '':return Nonereturn data#通过获取关键字拿到 searchjson data数据def get_data_for_json(self,row):opera_json = OperationJson()request_data = opera_json.get_data(self.get_request_data(row))if request_data == '':return Nonereturn request_data#获取预期结果def get_expect_data(self,row):col = int(data_config.get_expect())expect = self.opera_excel.get_cell_value(row,col)if expect == '':return Nonereturn expectdef write_result(self, row, value):col = int(data_config.get_result())self.opera_excel.write_value(row, col, value)#判断是否有case依赖def is_depend(self,row):col = int(data_config.get_case_dependid())depend_case_id = self.opera_excel.get_cell_value(row,col)if depend_case_id == "":return Noneelse:return depend_case_id# 获取依赖数据的keydef get_depend_key(self, row):col = int(data_config.get_data_depend())depend_key = self.opera_excel.get_cell_value(row, col)if depend_key =="":return Noneelse:return depend_key#获取数据依赖字段def get_depend_field(self,row):col = int(data_config.get_field_depend())data = self.opera_excel.get_cell_value(row,col)if data =="":return Noneelse:return data

data_config.py

封装常量：

#coding:utf-8#封装获取常量class global_var:#case_idId = '0'request_name = '1'url = '2'run = '3'request_way = '4'header = '5'case_dependid = '6'data_depend = '7'field_depend = '8'data = '9'expect = '10'result = '11'#获取caseiddef get_id():return global_var.Iddef get_request_name():return global_var.request_namedef get_url():return global_var.urldef get_run():return global_var.rundef get_request_way():return global_var.request_waydef get_header():return global_var.headerdef get_case_dependid():return global_var.case_dependiddef get_data_depend():return global_var.data_dependdef get_field_depend():return global_var.field_dependdef get_data():return global_var.datadef get_expect():return global_var.expectdef get_result():return global_var.resultclass Getheadervalue():def __init__(self,header):self.header = headerdef get_header_value(seif,header):return {"Content-Type":"application/json;charset=UTF-8"}

operation_excel

操作excel表格，获取数据

#coding:utf-8import xlrdfrom xlutils.copy import copyclass OperationExcel:def __init__(self,file_name=None,sheet_id=None):if file_name:self.file_name = file_nameself.sheet_id = sheet_idelse:self.file_name = 'D:\daima\ORSD\dataconfig\interface.xls'self.sheet_id = 0self.data = self.get_data()#获取sheets的内容def get_data(self):data = xlrd.open_workbook(self.file_name)tables = data.sheets()[self.sheet_id]return tables# 获取单元格的行数def get_lines(self):tables = self.datareturn tables.nrows#获取某一个单元格的内容def get_cell_value(self,row,col):return self.data.cell_value(row,col)#写入excel数据def write_value(self,row,col,value):read_data = xlrd.open_workbook(self.file_name)write_data = copy(read_data)sheet_data = write_data.get_sheet(0)sheet_data.write(row,col,value)write_data.save(self.file_name)#根据对应的caseid 找到对应行的内容def get_rows_data(self,case_id):row_num = self.get_row_num(case_id)rows_data = self.get_row_values(row_num)return rows_data#根据对应的caseid 找到对应的行号def get_row_num(self,case_id):num = 0clols_data = self.get_cols_data()for col_data in clols_data:if case_id in col_data:return numnum = num + 1#根据对应行号，找到该行的内容def get_row_values(self,row):tables = self.datarow_data = tables.row_values(row)return row_data#根据某一个获取某一列的内容def get_cols_data(self,col_id=None):if col_id != None:cols = self.data.col_values(col_id)else:cols = self.data.col_values(0)return colsif __name__ == '__main__':opers = OperationExcel()print opers.get_cell_value()

operation_header.py

获取token:

#coding:utf-8import requestsimport jsonfrom operation_json import OperationJson'''写入header 中的 token，在token.json文件中'''class OperationHeader:def __init__(self,response):self.res = json.loads(response)def get_response_token(self):'''获取登录返回的accessToken'''token1 = self.res['data']['accessToken']token = {"Authorization":token1}return tokendef write_token(self):token = self.get_response_token()op_json = OperationJson()op_json.write_data(token)if __name__ == '__main__':op_header = OperationHeader(res)op_header.write_token()

run_main.py

#coding:utf-8import syssys.path.append("D:/daima/ORSD")from base.runmethod import RunMethodfrom data.get_data import GetDatafrom mon_util import CommonUtilfrom data.dependdata import DependentDatafrom util.operation_header import OperationHeaderfrom util.operation_json import OperationJsonimport jsonfrom data.data_config import Getheadervaluefrom util.send_email import SendEmailimport HTMLTestRunnerclass RunTest:def __init__(self):self.run_method = RunMethod()self.data = GetData()_util = CommonUtil()self.send_email = SendEmail()# 程序执行的def test_go_on_run(self):res = Nonepass_count = []fail_count = []#10 0,1,2,3rows_count = self.data.get_case_lines()for i in range(1,rows_count):is_run = self.data.get_is_run(i)if is_run:url = self.data.get_request_url(i)method = self.data.get_request_method(i)request_data = self.data.get_data_for_json(i)expect = self.data.get_expect_data(i)header = self.data.is_header(i)depend_case = self.data.is_depend(i)if depend_case != None:self.depend_data = DependentData(depend_case)# 获取的依赖的返回数据depend_response_data = self.depend_data.get_data_for_key(i)# 获取依赖字段depend_key = self.data.get_depend_field(i)request_data2 = {depend_key: depend_response_data} # 请求数据=依赖的返回数据request_data = json.dumps(request_data2)if header == "yes":op_json = OperationJson('D:\daima\ORSD\dataconfig/token.json')token = op_json.get_data('Authorization')header = {"Content-type": "application/json;charset=UTF-8","Authorization": 'Bearer ' + eval(token)}res = self.run_method.run_main(method,url,request_data,header)if header == "header":header = Getheadervalue(header)header = header.get_header_value(header)res = self.run_method.run_main(method,url,request_data,header)op_header = OperationHeader(res)op_header.write_token()elif header == "yes":op_json = OperationJson('D:\daima\ORSD\dataconfig/token.json')token = op_json.get_data('Authorization')header = {"Content-type": "application/json;charset=UTF-8","Authorization": 'Bearer ' + eval(token)}res = self.run_method.run_main(method,url,request_data,header)else:res = self.run_method.run_main(method,url,request_data,header)# print resif _util.is_contain(expect,res):self.data.write_result(i,'pass')pass_count.append(i)else:self.data.write_result(i,res)fail_count.append(i)self.send_email.send_main(pass_count,fail_count)print len(pass_count)print len(fail_count)# return resif __name__ == '__main__':run = RunTest()run.test_go_on_run()

send_email.py

#coding:utf-8import smtplib #发送邮件from email.mime.text import MIMEText #邮件格式class SendEmail:global send_userglobal email_hostglobal passwordemail_host = ""send_user = "xxxxxxxxxxx@"password = "xxxxxxxxx"def send_mail(self,user_list,sub,content):user = "why"+"<"+send_user+">"message = MIMEText(content,_subtype='plain',_charset='utf-8')message['Subject'] = submessage['From'] = usermessage['To'] = ";".join(user_list)server = smtplib.SMTP()server.connect(email_host)server.login(send_user,password)server.sendmail(user,user_list,message.as_string())server.close()def send_main(self,pass_list,fail_list):pass_num = float(len(pass_list))fail_num = float(len(fail_list))count_num = pass_num+fail_num#90%pass_result = "%.2f%%" %(pass_num/count_num*100)fail_result = "%.2f%%" %(fail_num/count_num*100)user_list = ['xxxxxxxxxn@']sub = "接口自动化报告"content = "此次一共运行接口个数为%s个，通过个数为%s个，失败个数为%s,通过率为%s,失败率为%s" %(count_num,pass_num,fail_num,pass_result,fail_result )self.send_mail(user_list,sub,content)if __name__ == '__main__':sen = SendEmail()sen.send_main([1,2,3,4],[2,3,4,5,6,7])

DependentData 方法：

class DependentData:def __init__(self,case_id):self.case_id = case_idself.opera_excel = OperationExcel()self.data = GetData()self.run_method = RunMethod()#通过case_id 获取该case_id的整行数据def get_case_line_data(self,case_id):rows_data = self.opera_excel.get_rows_data(self.case_id)return rows_data#执行依赖测试获取结果def run_dependent(self):run_method = RunMethod()row_num = self.opera_excel.get_row_num(self.case_id)request_data = self.data.get_data_for_json(row_num)method = self.data.get_request_method(row_num)url = self.data.get_request_url(row_num)header = self.data.is_header(row_num)if header == "header":header = Getheadervalue(header)header = header.get_header_value(header)res = self.run_method.run_main(method, url, request_data, header)op_header = OperationHeader(res)op_header.write_token()return json.loads(res)#根据依赖的字段去获取执行依赖测试case的响应数据然后返回def get_data_for_key(self,row):depend_data = self.data.get_depend_key(row) #依赖的返回数据response_data = self.run_dependent() #执行依赖测试json_exe = parse(depend_data)madle = json_exe.find(response_data)return [math.value for math in madle][0]