Skip to content

A
apiTest1.0
Commit 85f961b1 by zhangchunyao
Options

code commit 2.0 201907311711

parent 6cca67d8
Showing with 367 additions and 206 deletions
case/testapi.py
"""
by LiangChao
at 2019/7/31
用来执行单接口用例文件
"""
from common.Http import Http
from common.Excel import Reader, Writer
import unittest
import json,ddt
import json, ddt, textwrap
http = Http()
reader = Reader()
writer = Writer()
#reader.open_excle('D:/workSpace/Python_workspace/testcase.xls')
reader.open_excle('/opt/apitest/apiAutoByJenkins/repot/testcase.xls')
# 读取用例文件
# reader.open_excle('D:/workSpace/Python_workspace/apitest2.0.xls') # for local
reader.open_excle('/opt/apitest/apiAutoByJenkins/repot/apitest.xls') # for server
sheetname = reader.get_sheets()
#writer.copy_open('D:/workSpace/Python_workspace/testcase.xls', 'D:/workSpace/Python_workspace/testcase1.xls')
writer.copy_open('/opt/apitest/apiAutoByJenkins/repot/testcase.xls', '/opt/apitest/apiAutoByJenkins/repot/testcase1.xls')
# copy用例文件
# writer.copy_open('D:/workSpace/Python_workspace/apitest2.0.xls', 'D:/workSpace/Python_workspace/apitest2.0_1.xls') # for local
writer.copy_open('/opt/apitest/apiAutoByJenkins/repot/apitest.xls', '/opt/apitest/apiAutoByJenkins/repot/apitest1.xls') # for server
sheetname1 = writer.get_sheets()
writer.set_sheet(sheetname1[0])
for sheet in sheetname:
......@@ -23,62 +29,58 @@ class Run(unittest.TestCase):
@classmethod
def setUp(self):
pass
def runner(self,line,http,num):
def runner(self,line,http):
d = {
"Post": [line[4], line[5], line[6], line[7], line[8]],
"Addheader": [line[4]],
"savejson": [line[4], line[5]],
"clear_values": [],
"Url": [line[4]],
"Addjson": [line[4], line[5]],
"Database_query": [line[4], line[5], line[6]],
"Get": [line[4], line[5], line[6], line[7], line[8]]
}
if len(line[0])>0 or len(line[1])>0:
return
if line[3] == 'post':
result = http.post(line[4], line[5], line[6])
return result
if line[3] == 'addheader':
result = http.addheader(line[4])
return result
if line[3] == 'assertquals':
result = http.assertquals(line[4], line[5])
if result[1] == 'fail':
writer.write(num, 7, result[1])
try:
self.assertEqual(result[0][0], result[0][1])
except AssertionError as e:
print("预期与实际不符:{}".format(e))
raise
return result[1]
if line[3] == 'assertquals_int':
result = http.assertquals_int(line[4], line[5])
if result[1] == 'fail':
writer.write(num, 7, result[1])
try:
self.assertEqual(result[0][0], result[0][1])
except AssertionError as e:
print("预期与实际不符:{}".format(e))
raise
return result[1]
if line[3] == 'savejson':
result = http.savejson(line[4], line[5])
return result
if line[3] == 'clear_values':
http.clear_values()
return
if line[3] == 'seturl':
result = http.seturl(line[4])
return result
if line[3] == 'Addjson':
result = http.Addjson(line[4])
return result
if line[3] == 'Database_query':
result = http.Database_query(line[4],line[5])
return result
if line[3] =='get':
result = http.get(line[4], line[5], line[6])
return result
if hasattr(http, line[3]):
func = getattr(http, line[3])
for k, v in d.items():
if line[3] in k:
result = func(*v)
return result
@ddt.data(*data)
def test_run(self,data):
result = self.runner(reader.readline_new(data[-1]), http, data[-1])
# 用例执行
excle_value = reader.readline_new(data[-1])
result = self.runner(excle_value, http)
if result != None:
#print(result)
results = json.dumps(result)
writer.write(data[-1], 7, results)
if isinstance(result, tuple):
results = json.dumps(result[0])
if result[1][0] == result[1][1]:
writer.write(data[-1], 9, 'pass')
else:
writer.write(data[-1], 9, 'fail')
incol = 10
for var in textwrap.wrap(text=results, width=30000):
writer.write(data[-1], incol, var)
incol = incol +1
try:
self.assertEqual(result[1][0], result[1][1], msg='\n检验的字段名称:{}\n请求的接口api:{}'.format(excle_value[7], excle_value[4]))
except AssertionError as e:
print("实际与预期不符:{}\n接口返回结果:\n{}".format(e, result[0]))
#print("实际与预期不符:{}\n接口返回结果:\n".format(e))
raise
print(result[0])
# 写入Excel文件
else:
print(result)
results = json.dumps(result)
# 写入Excel文件
writer.write(data[-1], 9, 'pass')
writer.write(data[-1], 10, results)
#writer.save_close()
def tearDown(self):
# 保存文件
writer.save_close()
if __name__ == '__main__':
......
case/testcase.py
"""
by LiangChao
at 2019/7/31
用来执行多接口流程用例文件
"""
from common.Http import Http
from common.Excel import Reader, Writer
from common.Http import InterFace
from xlutils.copy import copy
import unittest
import json
import ddt
import json, ddt, textwrap
http = Http()
reader = Reader()
writer = Writer()
inter = InterFace()
APIPATH = '/opt/apitest/apiAutoByJenkins/repot/apitest.xls'
#APIPATH = 'D:/workSpace/Python_workspace/apitestunit.xls'
br = reader.open_excle(APIPATH)
# 读取用例文件
# reader.open_excle('D:/workSpace/Python_workspace/testcase2.0.xls') # for local
reader.open_excle('/opt/apitest/apiAutoByJenkins/repot/testcase.xls') # for server
sheetname = reader.get_sheets()
bw = copy(br)
data = reader.sheet_cont()
# copy用例文件
# writer.copy_open('D:/workSpace/Python_workspace/testcase2.0.xls', 'D:/workSpace/Python_workspace/testcase2.0_1.xls') # for local
writer.copy_open('/opt/apitest/apiAutoByJenkins/repot/testcase.xls', '/opt/apitest/apiAutoByJenkins/repot/testcase1.xls') # for server
sheetname1 = writer.get_sheets()
writer.set_sheet(sheetname1[0])
for sheet in sheetname:
# 设置当前读取的sheet页面
reader.set_sheet(sheet)
data = reader.dict_data()
#print(data)
@ddt.ddt
class Run(unittest.TestCase):
@classmethod
def setUp(self):
pass
def runner(self,line,http):
d = {
"Post": [line[4], line[5], line[6], line[7], line[8]],
"Addheader": [line[4]],
"savejson": [line[4], line[5]],
"clear_values": [],
"Url": [line[4]],
"Addjson": [line[4], line[5]],
"Database_query": [line[4], line[5], line[6]],
"Get": [line[4], line[5], line[6], line[7], line[8]]
}
if len(line[0])>0 or len(line[1])>0:
return
if hasattr(http, line[3]):
func = getattr(http, line[3])
for k, v in d.items():
if line[3] in k:
result = func(*v)
return result
@ddt.data(*data)
def test_run(self,data):
result = inter.dosheet(br, bw, data[-1])
#print(result[0])
if result[1]:
if isinstance(result[1][0], int):
# 用例执行
excle_value = reader.readline_new(data[-1])
result = self.runner(excle_value, http)
if result != None:
if isinstance(result, tuple):
results = json.dumps(result[0])
if result[1][0] == result[1][1]:
writer.write(data[-1], 9, 'pass')
else:
writer.write(data[-1], 9, 'fail')
incol = 10
for var in textwrap.wrap(text=results, width=30000):
writer.write(data[-1], incol, var)
incol = incol +1
try:
self.assertEqual(result[1][0], int(result[1][1]), msg='\n检验的字段名称:{}\n请求的接口api:{}'.format(result[1][-2], result[1][-1]))
self.assertEqual(result[1][0], result[1][1], msg='\n检验的字段名称:{}\n请求的接口api:{}'.format(excle_value[7], excle_value[4]))
except AssertionError as e:
print("实际与预期不符:{}\n接口返回结果:\n{}".format(e, result[0]))
#print("实际与预期不符:{}\n接口返回结果:\n".format(e))
raise
print(result[0])
# 写入Excel文件
else:
try:
self.assertEqual(result[1][0], result[1][1], msg='\n检验的字段名称:{}\n请求的接口api:{}'.format(result[1][-2], result[1][-1]))
except AssertionError as e:
print("实际与预期不符:{}\n接口返回结果:\n{}".format(e, result[0]))
raise
print(result)
results = json.dumps(result)
# 写入Excel文件
writer.write(data[-1], 9, 'pass')
writer.write(data[-1], 10, results)
#writer.save_close()
def tearDown(self):
inter.xlsx_save(bw, APIPATH)
# 保存文件
writer.save_close()
if __name__ == '__main__':
run = Run()
......
common/Excel.py
'''
by LiangChao
at 2019/7/11
用来读/写Excel文件内容
'''
import os,xlrd
from xlutils.copy import copy
class Reader:
'''
by LiangChao
at 2019/7/11
用来读取Excel文件内容
'''
def __init__(self):
#整个excel工作簿缓存
self.workbook=None
......@@ -82,7 +83,6 @@ class Reader:
r.append(values)
return r
def sheet_cont(self,):
r = []
for i in list(range(self.rows-1)):
......@@ -119,10 +119,10 @@ class Reader:
self.set_sheet(sheet)
for i in range(self.rows):
line = self.readline()
if line[10] == 'fail':
api.append(line[4])
if line[9] == 'fail':
result = line[4]+":"+line[2]
api.append(result)
return api
class Writer:
'''
by kongwen 2019/7/8 用来复制写入excel
......@@ -140,7 +140,7 @@ class Writer:
self.row=0
#记录写入的列
self.clo=0
#复制并打开excel
# 复制并打开excel
def copy_open(self,srcfile,dstfile):
#判断要复制的文件是否存在
if not os.path.isfile(srcfile):
......@@ -159,22 +159,22 @@ class Writer:
#默认使用第一个sheet
#sheet=wb.get_sheet('Sheet1')
return
#获取sheet页面
# 获取sheet页面
def get_sheets(self):
#获取所有sheet的名字，并返回第一个列表
sheets=self.workbook.sheet_names()
#print(sheets)
return sheets
#切换sheet页面
# 切换sheet页面
def set_sheet(self,name):
#通过sheet名字，切换sheet页面
self.sheet=self.wb.get_sheet(name)
return
#写入指定单元格，保留原格式
# 写入指定单元格，保留原格式
def write(self,r,c,value):
#获取写入的单元格
def _getCell(sheet,r,c):
#获取行
# 获取写入的单元格
def _getCell(sheet, r, c):
# 获取行
row=sheet._Worksheet__rows.get(r)
if not row:
return None
......@@ -190,32 +190,33 @@ class Writer:
ncell=_getCell(self.sheet,r,c)
if ncell:
#设置写入后格式和写入前一样
ncell.xf_idx=cell.xf_idx
ncell.xf_idx = cell.xf_idx
return
#保存
def save_close(self):
#保存复制后的文件到硬盘
self.wb.save(self.df)
return
#调试
if __name__ == '__main__':
reader=Reader()
reader.open_excle('/Users/niweizhong/Desktop/case1.xls')
reader.open_excle('D:/dminterfacetest/repot/apitest.xls')
sheetname=reader.get_sheets()
api = []
for sheet in sheetname:
#设置当前读取的sheet页面
reader.set_sheet(sheet)
for i in range(reader.rows):
print(reader.readline())
line = reader.readline()
if line[10] == 'fail':
api.append(line[4])
print(api)
writer=Writer()
writer.copy_open('/Users/niweizhong/Desktop/case1.xls','/Users/niweizhong/Desktop/case2.xls')
sheetname=writer.get_sheets()
writer.set_sheet(sheetname[0])
writer.write(1,1,'William')
writer.save_close()
# writer=Writer()
# writer.copy_open('/Users/niweizhong/Desktop/case1.xls','/Users/niweizhong/Desktop/case2.xls')
# sheetname=writer.get_sheets()
# writer.set_sheet(sheetname[0])
# writer.write(1,1,'William')
# writer.save_close()
common/Http.py
'''
by LiangChao
at 2019/7/11
接口运行文件
'''
import requests
import json,time, xlrd, xlwt, textwrap
import json, time, xlrd, xlwt, textwrap
import jsonpath
import pymysql
from urllib.parse import urlencode
from common.Excel import Reader
class Http():
'''
by LiangChao
......@@ -19,15 +25,10 @@ class Http():
'Hm_lvt_47362f54f594efad07774d01ad0858d5': '1559783995,1560069495,1560079873,1560735612',
'web_login_cookie_id': 'fefd1c81641711e69d0818c58a146fd2',
'Hm_lpvt_47362f54f594efad07774d01ad0858d5': '1560844337'}
self.value = []
self.value = {}
self.header = {}
# def addheader(self,heaad):
# heaads = json.loads(heaad)
# for key in heaads.keys():
# self.header[key] = heaads[key]
# return self.header
def addheader(self,header):
def Addheader(self,header):
"""添加头部信息"""
headers = json.loads(header)
# 循环取headers中value，存放至self.header中
......@@ -41,74 +42,77 @@ class Http():
def addparam(self,paramters='',str=''):
"""添加参数"""
if paramters =='':
paramters ='{}'
paramters = '{}'
self.params = json.loads(paramters)
# 针对transId与timestamp参数进行时间格式化
for i in self.params.keys():
if i == 'transId' or i == 'timestamp':
a = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
self.params[i] = self.params[i] + a
#self.params = paramters
# self.params = paramters
if str == '':
return self.params
params = self.spelling_parameters(self.params)
return params
else:
key = json.loads(str)
keylist = key['params']
i = 0
value_list = self.value
for k in keylist:
self.params[k] = value_list[i]
i = i+1
#print(self.params)
return self.params
def seturl(self,url):
for k in key.keys():
try:
self.params[k] = self.value[k]
except:
print('参数不存在:{}'.format(self.params[k]))
params = self.spelling_parameters(self.params)
return params
def spelling_parameters(self,parameters):
for p in parameters.keys():
if not type(parameters[p]) in [str]:
parameters[p] = str(parameters[p])
return parameters
def Url(self,url):
"""设置URL"""
self.url = url
return self.url
# def post(self, api, default, variable=''):
# param = self.addparam(default, variable)
# if 'application/jso' in self.header:
# param1 = json.dumps(param)
# self.result = self.session.post(self.url + api,headers=self.header, data=param1, cookies=self.cookies)
# self.header['Content-Type'] = 'application/x-www-form-urlencoded'
# else:
# self.result = self.session.post(self.url + api, data=param, cookies=self.cookies)
# #self.cookies = self.result.cookies
# self.jsons = json.loads(self.result.text)
# #self.value = []
# return self.jsons
def post(self, api, default, variable=''):
def Post(self, api, default, variable='', check='', expected=''):
"""post请求"""
# 调用self.addparam方法，生成入参参数
param = self.addparam(default, variable)
#print(param)
if self.header != {}:
# if 'json' in self.header:
# param1 = json.dumps(param)
self.result = self.session.post(self.url + api, headers=self.header, data=param, cookies=self.cookies)
self.header = {}
if 'application/json' in self.header["Content-Type"]:
param = json.dumps(param)
self.result = self.session.post(self.url + api, headers=self.header, data=param)
self.header.clear()
else:
# 进行post请求，获取返回信息
self.result = self.session.post(self.url + api, data=param, cookies=self.cookies)
self.result = self.session.post(self.url + api, data=param)
#self.cookies = self.result.cookies
self.jsons = json.loads(self.result.text)
return self.jsons
arr = self.assertquals(check, expected)
return self.jsons, arr
def get(self,api,default, variable=''):
def Get(self, api, default, variable='', check='', expected=''):
"""get请求"""
param = self.addparam(default, variable)
#print(param)
paramts = urlencode(param)
# 进行get请求，获取返回信息
self.result = self.session.get(self.url + api + paramts)
#self.cookies = self.result.cookies
# self.cookies = self.result.cookies
self.jsons = json.loads(self.result.text)
return self.jsons
def timing(self,n):
arr = self.assertquals(check, expected)
return self.jsons, arr
def Time(self,n):
"""定时休眠"""
time.sleep(n)
def savejson(self,key,value):
results = self.jsons[value]
self.params[key] = results
return self.params
def savecookies(self):
self.cookiess = self.cookies
return self.cookiess
......@@ -116,18 +120,17 @@ class Http():
def assertquals(self,key,values):
"""断言str类型"""
dic = self.jsons
value = jsonpath.jsonpath(dic, key)
value.append(values)
if value[0] == value[1]:
return value, 'pass'
else:
return value, 'fail'
# if value[0] == values:
# print('pass')
# else:
# print('fail')
try:
value = jsonpath.jsonpath(dic, key)
if type(value[0]) in [int]:
value.append(int(values))
else:
value.append(values)
return value
except:
print('未在返回结果取到值:{}'.format(dic))
def assertquals_int(self,key,values):
def assertquals_int(self, key, values):
"""断言int类型"""
dic = self.jsons
value = jsonpath.jsonpath(dic, key)
......@@ -136,11 +139,6 @@ class Http():
return value, 'pass'
else:
return value, 'fail'
# print(values)
# if value[0] == int(values):
# return 'pass'
# else:
# return 'fail'
def __getparms(self,str):
for key in self.params.keys():
......@@ -193,12 +191,19 @@ class Http():
#print(value)
return value
# jsonpath 取值
def Addjson(self, locators):
def Addjson(self, locators, parameters):
"""jsonpath取值"""
dic = self.jsons
value = jsonpath.jsonpath(dic, locators)
#print(value)
self.value.append(value[0])
try:
value = jsonpath.jsonpath(dic, locators)
# print(value)
param = json.loads(parameters)
for p in param.keys():
self.value[p] = param[p]
self.value[p] = value[0]
except:
print('未在返回结果取到值:{}'.format(dic))
raise
return value
def can_convert_to_int(self,input):
......@@ -210,15 +215,12 @@ class Http():
# 清除self.value内容，用于初始化
# def clear_values(self):
# self.value = []
# return self.value
def clear_values(self):
self.value.clear()
return self.value
# 数据库取值
def Database_query(self,dbname,sql):
def Database_query(self,dbname,sql, param):
conn = pymysql.connect(host="56cbb9f62fac6.sh.cdb.myqcloud.com",
user="cdb_outerroot",
password="@09ui%sbc09",
......@@ -235,14 +237,17 @@ class Http():
cursor.close()
# 关闭数据库连接
conn.close()
for key in res[0]:
self.value.append(key)
print(param)
params = json.loads(param)
for p in params.keys():
self.value[p] = params[p]
self.value[p] = res[0][0]
return self.value
class InterFace():
''''''
def __init__(self):
self.session = requests.session()
# 打开文件
def xlsx_open(slef,filepath):
book = xlrd.open_workbook(filepath)
......@@ -275,29 +280,33 @@ class InterFace():
#params1 = object["params"]
#print(params1)
if object["headers"]!="":
if 'application/json' in object["headers"]:
dic = json.loads(object["params"])
if 'application/jso' in object["headers"]:
params1 = json.dumps(dic)
#params1 = json.dumps(object["params"])
#print(params1)
head=object["headers"]
header=json.loads(head)
response = slef.session.post(object["url"], params1,headers=header)
response = slef.session.post(object["url"], params1, headers=header)
a = response.status_code()
print(a)
#response = requests.post(object["url"], params1,headers=header,cookies=cookies)
results = response.json()
else:
response = slef.session.post(object["url"], params1)
response = slef.session.post(object["url"], data=params1)
a = response.status_code
print(a)
#response = requests.post(object["url"], params1,cookies=cookies)
results = response.json()
result = results
elif object["method"] == "get":
print(object["params"])
#print(object["params"])
dic = json.loads(object["params"])
params1 = urlencode(dic)
response = slef.session.get(object["url"]+params1)
#print(response)
#response = requests.get(object["url"], params1)
#result = json.loads(response.text)
result = response.json()
else:
return 0
return result
......@@ -320,7 +329,7 @@ class InterFace():
slef.xlsx_set(bwt_sheet, inrow, incol, var, False)
incol = incol + 1
def dosheet(slef,brd, bwt, i):
def dosheet(slef, brd, bwt, i):
brd_sheet = brd.sheets()[0]
bwt_sheet = bwt.get_sheet(0)
object = slef.xlsx_getRow(brd_sheet, i)
......@@ -348,5 +357,4 @@ class InterFace():
arr.append(object["expected_code"])
arr.append(object["code"])
arr.append(object["path"])
return result, arr
return result, arr
\ No newline at end of file
common/resultToDing.py 0 → 100644
# coding: UTF-8
'''
pip install DingtalkChatbot
'''
# 2019/7/22 created by dadu
from dingtalkchatbot.chatbot import DingtalkChatbot
import time
from common.Excel import Reader
import sys,os
runTime = time.strftime('%Y-%m-%d %H:%M:%S')
#created by 2019/7/24
def findModuleName(arr,str):
for index in range (len(arr)):
if str == arr[index]:
return 1
return 0
#https://oapi.dingtalk.com/robot/send?access_token=1ffb37a991040d9d58bc3e967fa563e379db22c448fd7440f747cd4c44d4db26
def findGroupDingAddr(str):
enterGroupModule = ["gic-store-service","gic-store-web","gic-store-web"]
memberGroupModule = ["g2Test1","g2Test2"]
marketGroupModule = ["api-open"]
haobanGruupModule = ["g4Test1","g4Test2"]
enterGroupDingAddr = 'https://oapi.dingtalk.com/robot/send?access_token=1ffb37a991040d9d58bc3e967fa563e379db22c448fd7440f747cd4c44d4db26'
memberGroupDingAddr = 'https://oapi.dingtalk.com/robot/send?access_token=2656a52b44f1e2d9ebbf5e57560b5a7c2619dc07cec35c62d19b8b252e24a064'
marketGroupDingAddr = 'https://oapi.dingtalk.com/robot/send?access_token=3e8fac3b110012fa44ae42235323deba5102fc439b62d87b7491c838b1948470'
haobanGruupDingAddr = 'https://oapi.dingtalk.com/robot/send?access_token=2656a52b44f1e2d9ebbf5e57560b5a7c2619dc07cec35c62d19b8b252e24a064'
err = "no group can be match,please check it"
if findModuleName(enterGroupModule,str) == 1:
return enterGroupDingAddr
elif findModuleName(memberGroupModule,str) == 1:
return memberGroupDingAddr
elif findModuleName(marketGroupModule,str) == 1:
return marketGroupDingAddr
elif findModuleName(haobanGruupModule,str) == 1:
return haobanGruupDingAddr
else :
print(err)
return 0
def resultTextFalse(reporturl,str):
# path = 'D:/workSpace/Python_workspace/apitest2.0_1.xls' # for local
path = '/opt/apitest/apiAutoByJenkins/repot/apitest1.xls' # for server
if os.path.exists(path):
readExp = Reader()
arrExpSingle = readExp.api_result(path)
# print(arrExpSingle)
expMsg = '\n'.join(arrExpSingle)
if(findGroupDingAddr(str)!=0):
groupDing = DingtalkChatbot(findGroupDingAddr(str))
groupDing.send_text(msg = "接口自动化执行异常结果\n执行时间："+ runTime+ "\n执行异常接口列表：\n"+expMsg+"\n详细结果地址："+reporturl)
else:
sys.exit(0)
else :
print("resultFile is not exist,please check it")
sys.exit(0)
def resultTextPass(reporturl,str):
if(findGroupDingAddr(str)!=0):
groupDing = DingtalkChatbot(findGroupDingAddr(str))
groupDing.send_text(msg = "接口自动化执行无误\n执行时间："+ runTime+ "\n详细结果地址："+reporturl)
else :
sys.exit(0)
def msgSend(passrate,reporturl,str):
#print(str)
if float(passrate)==100.00:
resultTextPass(reporturl,str)
else :
resultTextFalse(reporturl,str)
common/resultToDingForQateam.py
......@@ -3,19 +3,27 @@
from dingtalkchatbot.chatbot import DingtalkChatbot
import time
from common.Excel import Reader
import sys
import sys,os
qaTeamWebhook = 'https://oapi.dingtalk.com/robot/send?access_token=2656a52b44f1e2d9ebbf5e57560b5a7c2619dc07cec35c62d19b8b252e24a064'
qaTeamDing = DingtalkChatbot(qaTeamWebhook)
runTime = time.strftime('%Y-%m-%d %H:%M:%S')
def resultTextFalse(reporturl):
#path = 'D:/workSpace/Python_workspace/apitestunit.xls'
path = '/opt/apitest/apiAutoByJenkins/repot/apitest.xls'
readExp = Reader()
arrExpSingle = readExp.api_result(path)
expMsg = '\n'.join(arrExpSingle)
qaTeamDing.send_text(msg = "接口自动化执行异常结果\n执行时间："+ runTime+ "\n执行异常接口列表：\n"+expMsg+"\n详细结果地址："+reporturl)
# pathSingle = 'D:/workSpace/Python_workspace/apitest2.0_1.xls' # for local
# pathMulti = 'D:/workSpace/Python_workspace/testcase2.0_1.xls' # for local
pathSingle = '/opt/apitest/apiAutoByJenkins/repot/apitest1.xls' # for server
pathMulti = '/opt/apitest/apiAutoByJenkins/repot/testcase1.xls' #for server
if (os.path.exists(pathSingle) or os.path.exists(pathMulti)):
readExp = Reader()
arrExpSingle = readExp.api_result(pathSingle)
arrExpMulti = readExp.api_result(pathMulti)
expSingleMsg = '\n'.join(arrExpSingle)
expMultiMsg = '\n'.join(arrExpMulti)
qaTeamDing.send_text(msg = "接口自动化执行异常结果\n执行时间："+ runTime+ "\n执行异常接口列表：\n"+expSingleMsg+"\n"+expMultiMsg+"\n详细结果地址："+reporturl)
else :
print("resultFile is not exist,please check it")
qaTeamDing.send_text(msg = "接口自动化执行异常结果\n执行时间："+ runTime+ "\n详细结果地址："+reporturl)
def resultTextPass(reporturl):
qaTeamDing.send_text(msg = "接口自动化执行无误\n执行时间："+ runTime+ "\n详细结果地址："+reporturl)
......
runner.py
# encoding: utf-8
import unittest
import HTMLTestReportCN
from common.Excel import Reader
import time
from common.Mail import sendResultMail
from HTMLTestReportCN import TestResult
#from common.resultToDing import msgSend
# from common.resultToDing import msgSend
from common.resultToDingForQateam import msgSend
import sys
#case_dir = 'D:/workSpace/Python_workspace/myPython/.idea/interfaceTest_ddt/case'
#report_dir = 'D:/workSpace/Python_workspace/myPython/.idea/interfaceTest_ddt/repot'
case_dir = '/opt/apitest/apiAutoByJenkins/case'
report_dir = '/opt/apitest/apiAutoByJenkins/repot'
'''
by LiangChao
at 2019/7/17
用来执行所有test开头的py文件，并生成表报
'''
reader = Reader()
# case_dir = 'D:/workSpace/Python_workspace/myPython/.idea/interfaceTest_ddt_2.0/case' #for local
# report_dir = 'D:/workSpace/Python_workspace/myPython/.idea/interfaceTest_ddt_2.0/repot' #for local
case_dir = '/opt/apitest/apiAutoByJenkins/case' # for server by QA
#case_dir = '/opt/apitest/apiAutoTest/case' # for server by DEV
report_dir = '/opt/apitest/apiAutoByJenkins/repot' # for server
discover = unittest.defaultTestLoader.discover(case_dir,pattern='test*.py')
#discover = unittest.defaultTestLoader.discover(case_dir,pattern='testcase.py')
# discover = unittest.defaultTestLoader.discover(case_dir,pattern='testcase.py')
# discover = unittest.defaultTestLoader.discover(case_dir,pattern='testapi.py')
now = time.strftime('%Y%m%d%H%M%S')
fileName = '/'+now+'test_report.html'
reportFileName = report_dir+fileName
report_url = "https://four.gicdev.com/apitest"
reportUrl = report_url+fileName
with open(reportFileName, 'wb') as f:
runner = HTMLTestReportCN.HTMLTestRunner(stream=f, title='接口自动化测试报告', description='接口测试用例执行情况', tester='jenkins')
runner.run(discover)
str=runner.getPassRate()
passRateValue = str.split('%',1)[0]
#msgSend(passRateValue,reportUrl,str)
#msgSend(passRateValue,reportUrl,sys.argv[1])
# api = reader.api_result('D:/workSpace/Python_workspace/apitest2.0_1.xls')
# print(api)
passRate = runner.getPassRate()
passRateValue = passRate.split('%',1)[0]
msgSend(passRateValue,reportUrl)
# msgSend(passRateValue,reportUrl,"gic-store-web")
# str = 'gic-store-web'
# msgSend(passRateValue,reportUrl,str)
# msgSend(passRateValue,reportUrl,sys.argv[1]) # for fangdong
msgSend(passRateValue,reportUrl) # for QaTeam
smtpSer ="smtp.exmail.qq.com"
# 测试服务器群发测试组成员
......@@ -41,3 +60,4 @@ sendResultMail(smtpSer,sendUser,sendPwd,recvUser,reportFileName)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment