Unittest框架+ddt数据驱动+HTMLTestRunner+sendmail（自动发送测试报告）+git+Jenkins

  1 import xlrd,os,copy
  2 
  3 import sys,os
  4 
  5 #这三行代码是解决文件路径问题
  6 
  7 curPath = os.path.abspath(os.path.dirname(__file__))
  8 
  9 rootPath = os.path.split(curPath)[0]
 10 
 11 sys.path.append(rootPath)
 12 
 13 class ExcelUtil():
 14 
 15     list=[]
 16 
 17     list1 = []
 18 
 19     def __init__(self,excelpath,sheetname=‘Sheet1‘):
 20 
 21         self.data=xlrd.open_workbook(excelpath)
 22 
 23         self.table=self.data.sheet_by_name(sheetname)
 24 
 25         #获取第一行作为key值
 26 
 27         self.keys=self.table.row_values(0)
 28 
 29         #获取总行数
 30 
 31         self.rowNum=self.table.nrows
 32 
 33         #获取总列数
 34 
 35         self.colNum=self.table.ncols
 36 
 37        
 38 
 39     #此方法有俩个用处：1、获取关闭或打开的case 2、不同模块的用例也可以通过该方法区分开来
 40 
 41     def open_case(self):
 42 
 43         exceldata = ExcelUtil(os.path.abspath(rootPath+‘/Case/Ddt_case_api/TestCase.xlsx‘), ‘Sheet1‘)
 44 
 45         list = []
 46 
 47         list1 = []
 48 
 49         list2=[]
 50 
 51         for i in exceldata.dict_data():
 52 
 53             a = i[‘control_case‘]  # 这是在Excel加的开关case的字段
 54 
 55             if a == ‘‘:
 56 
 57                 list.append(i)
 58 
 59             elif a==‘mobile_请先登录‘:
 60 
 61                 list1.append(i)
 62 
 63             elif a==‘B_请先登录‘:
 64 
 65                 list2.append(i)
 66 
 67         return  list,list1,list2
 68 
 69         # print(len(list))
 70 
 71     def dict_data(self):
 72 
 73         if self.rowNum<=1:
 74 
 75             print(‘总行数小于1‘)
 76 
 77         else:
 78 
 79             r=[]
 80 
 81             j=1
 82 
 83             for i in range(self.rowNum-1):
 84 
 85                 s={}
 86 
 87                 s[‘rowNum‘]=i+2
 88 
 89                 values=self.table.row_values(j)
 90 
 91                 for x in range(self.colNum):
 92 
 93                     s[self.keys[x]]=values[x]
 94 
 95                 r.append(s)
 96 
 97                 j+=1
 98 
 99             return r
100 
101 #通过下面这个入口进行调试
102 
103 if __name__==‘__main__‘:
104 
105     a=ExcelUtil(‘D:\Business_ManageMent\Case\Ddt_case_api\TestCase.xlsx‘)
106 
107     b=a.open_case()
108 
109     print(b)

 1 from openpyxl import load_workbook
 2 
 3 import openpyxl
 4 
 5 import xlrd
 6 
 7  
 8 
 9 def Copy_excel(exclepath1,excelpath2):
10 
11 #把excle1数据复制到excel2
12 
13     wb2=openpyxl.Workbook()
14 
15     wb2.save(excelpath2)
16 
17     wb1=load_workbook(exclepath1)
18 
19     wb2=load_workbook(excelpath2)
20 
21     sheets1=wb1.sheetnames
22 
23     sheets2=wb2.sheetnames
24 
25     sheet1=wb1[sheets1[0]]
26 
27     sheet2=wb2[sheets2[0]]
28 
29     maxrow=sheet1.max_row
30 
31     maxcolum=sheet1.max_column
32 
33     for m in range(1,maxrow+1):
34 
35         for n in range(97,97+maxcolum):
36 
37             n=chr(n)
38 
39             i=‘%s%d‘%(n,m)
40 
41             cell1=sheet1[i].value
42 
43             sheet2[i]=cell1
44 
45     wb2.save(excelpath2)
46 
47     wb1.close()
48 
49     wb2.close()
50 
51 class Write_excel():
52 
53     def __init__(self,filename):
54 
55         self.filename=filename
56 
57         self.wb=load_workbook(self.filename)
58 
59 #激活sheet
60 
61         self.ws=self.wb.active
62 
63     def write(self,row_n,col_n,value):
64 
65         self.ws.cell(row_n,col_n).value=value
66 
67         self.wb.save(self.filename)
68 
69  
70 
71  
72 
73 #通过下面入口进行调试
74 
75 if  __name__==‘__main__‘:
76 
77     Copy_excel(‘D:\Business_ManageMent\Case\Ddt_case_api\TestCase.xlsx‘,‘D:\Business_ManageMent\Case\\result.xlsx‘)
78 
79     wt=Write_excel(‘D:\Business_ManageMent\Case\\result.xlsx‘)
80 
81     wt.write(1,2,‘hello‘)

 1 import json
 2 import requests
 3 from Tool_class.read_excel_fz import ExcelUtil
 4 from Tool_class.writeexcel_fz import Copy_excel, Write_excel
 5 
 6 
 7 def send_requests(s, testdata):
 8     ‘‘‘封装requests请求‘‘‘
 9     #获取Excel标格中表头为method的数据
10     method = testdata["method"]
11     # 获取Excel标格中表头为url的数据
12     url = testdata["url"]
13     try:
14         #eval函数可以将读取到的参数内容转化成字典格式
15         # 获取Excel标格中表头为params的数据
16         params = eval(testdata["params"])
17     except:
18         params = None
19     # 请求头部headers
20     try:
21         # 获取Excel标格中表头为headers的数据
22         headers = eval(testdata["headers"])
23         print("请求头部：%s" % headers)
24     except:
25         headers = None
26 
27 
28     # post请求body类型
29     # 获取Excel标格中表头为type的数据
30     type = testdata["type"]
31     # 获取Excel标格中表头为id的数据
32     test_nub = testdata[‘id‘]
33     print("*******正在执行用例：-----  %s  ----**********" % test_nub)
34     print("请求方式：%s, 请求url:%s" % (method, url))
35     print("请求params：%s" % params)
36     # post请求body内容
37     try:
38         # 获取Excel标格中表头为body的数据
39          bodydata = eval(testdata["body"])
40     except:
41          bodydata = {}
42     # 判断传data数据还是json
43     if type == "json":
44             #json.dumps将字典数据转成json格式，因为不同的接口传递参数是不同的，有的是传data，有的传json
45             body= json.dumps(bodydata)
46     elif type == "data":
47             body = bodydata
48     else:
49             body = bodydata
50 
51     if type==‘json‘:
52         print("post请求body类型为：%s ,body内容为：%s" % (type,body))
53     elif method==‘post‘:
54         print("post请求body类型为：%s ,body内容为：%s" % (type,body))
55     verify = False
56     res = {}   # 接受返回数据
57 
58     try:
59         r = s.request(method=method,
60                       url=url,
61                       params=params,
62                       headers=headers,
63                       data=body,
64                       verify=verify
65                        )
66         print("页面返回信息：%s" % r.json())
67         res[‘id‘] = testdata[‘id‘]
68         res[‘rowNum‘] = testdata[‘rowNum‘]
69         res["statuscode"] = str(r.status_code)  # 状态码转成str
70         res["text"] = r.json()
71         res["times"] = str(r.elapsed.total_seconds())   # 接口请求时间转str
72         if res["statuscode"] != "200":
73             res["error"] = "服务错误，接口未请求成功"
74             res["msg"] = str(res["text"])
75         else:
76             res["error"] = ""
77             res["msg"] = ""
78         if testdata["checkpoint"] == res["text"][‘msg‘]:
79             res["result"] = "pass"
80             print("用例测试结果:   %s---->%s" % (test_nub, res["result"]))
81         else:
82             res["result"] = "fail"
83         return res
84     except Exception as msg:
85         res["msg"] = str(msg)
86         return res
87 #将运行返回的结果，根据自己需要，需要要哪些结果就把哪些结果写入拷贝的那份Excel中
88 def wirte_result(result, filename="D:\Business_ManageMent\Case\\result.xlsx"):
89     # 返回结果的行数row_nub
90     row_nub = result[‘rowNum‘]
91     # 写入statuscode
92     wt = Write_excel(filename)
93     wt.write(row_nub, 7, result[‘statuscode‘])        # 写入返回状态码statuscode,第8列
94     wt.write(row_nub, 12, result[‘times‘])             # 耗时
95     wt.write(row_nub, 14, result[‘error‘])            # 状态码非200时的返回信息
96     wt.write(row_nub, 13, result[‘result‘])           # 测试结果 pass 还是fail
97     wt.write(row_nub, 15, result[‘msg‘])              # 抛异常

 1 import time,os
 2 import smtplib
 3 import  unittest
 4 from Commons import HTMLTestRunner_jpg
 5 from email.header import Header
 6 from email.mime.text import MIMEText
 7 from email.mime.multipart import MIMEMultipart
 8 import email.mime.multipart
 9 from email.mime.application import MIMEApplication
10 from Case import *
11 
12 # test_dir = os.path.abspath(os.path.join(os.getcwd(), ".."))
13 # now = time.strftime(‘%y_%m_%d %H_%M_%S‘)
14 # filename = (os.path.abspath(‘../Report‘) + ‘\\‘ + now + ‘result.html‘)
15 # print(filename)
16 # fp = open(filename, ‘wb‘)
17 import sys,os
18 curPath = os.path.abspath(os.path.dirname(__file__))
19 rootPath = os.path.split(curPath)[0]
20 sys.path.append(rootPath)
21 #定义发送邮件
22 class Send():
23     def __init__(self):
24         self.flie_dir=rootPath+‘\Report‘
25         self.lists = os.listdir(self.flie_dir)
26         #将所有的报告按时间从小到大大排序
27         self.lists.sort(key=lambda fn: os.path.getmtime(self.flie_dir + ‘\\‘ + fn))
28         #取报告集合中最后一个报告即为最新的测试报告
29         self.new_file = os.path.join(self.flie_dir, self.lists[-1])
30     def send_mail(self):
31                 now=time.strftime(‘%y:%m:%d:%H:%M:%S‘)
32                 sender = ‘1063126729@qq.com‘
33                 recever= ‘steve@wemart.cn‘
34                 msg = MIMEMultipart()
35                 content = ‘最新接口测试报告，详情请下载附件查看，生成时间为:%s‘%(now)
36                 txt = email.mime.text.MIMEText(content, ‘plain‘, ‘utf-8‘)
37                 msg.attach(txt)
38                 msg[‘Subject‘]=‘接口自动化测试报告‘
39                 msg[‘date‘]=now
40                 #添加附件
41                 att = MIMEText(open(self.new_file, "rb").read(), "base64", "utf-8")
42                 att["Content-Type"] = "application/octet-stream"
43                 att["Content-Disposition"] = ‘attachment; filename= "Report.html"‘
44                 msg.attach(att)
45                 server=smtplib.SMTP_SSL(port=465)
46                 server.connect(‘smtp.qq.com‘)
47                 server.login(‘1063126729@qq.com‘,‘lhrcqszwzqafbcjf‘)
48                 server.sendmail(sender,recever,msg.as_string())
49                 server.quit()
50                 print(‘邮件已发送‘)
51     #创建一个测试套件，将所有的用例添加到测试套件
52     def creatsuit(slef):
53             test_dir =rootPath+‘\Case‘
54             suit=unittest.TestSuite()
55             discover=unittest.defaultTestLoader.discover(test_dir,pattern=‘test*.py‘,top_level_dir=None)
56             for test_suit in discover:
57                 for case in test_suit:
58                     suit.addTest(case)
59             return suit

 1 import unittest
 2 import ddt
 3 import os
 4 import requests
 5 from Commons import base_api
 6 from Tool_class import read_excel_fz
 7 from Tool_class import writeexcel_fz
 8 from Case.Hand_code_case.Login_case.Test_stor_login02 import *
 9 import os,copy
10 
11 #获取当前文件位置路径
12 curpath = os.path.dirname(os.path.realpath(__file__))
13 # 获取TestCase.xlsx路径
14 testxlsx = os.path.join(curpath, "TestCase.xlsx")
15 #获取curpath位置的上一层目录位置
16 report_path = os.path.join(os.path.dirname(curpath))
17 #获取测试结果表格的目录位置
18 reportxlsx = os.path.join(report_path, "result.xlsx")
19 #创建读取表格的对象
20 testdata = read_excel_fz.ExcelUtil(testxlsx)
21 #获取表格中需要运行的数据或不需要运行的数据
22 cases=testdata.open_case()
23 #表格中打开的case
24 case=cases[0]
25 #表格中关闭的case
26 case_mobile=cases[1]
27 
28 
29 @ddt.ddt
30 class Test_api(unittest.TestCase):
31       u‘‘‘B2C-API‘‘‘
32       @classmethod
33       def setUpClass(cls):
34           # 如果有登录的话，就在这里先登录了
35           cls.s = requests.Session()
36           # 复制xlsx
37           writeexcel_fz.Copy_excel(testxlsx, reportxlsx)
38 
39       #采用装饰器，在运行case之前都会先运行这个，这里的case是表格里打开的用例，也就是我需要运行的数据
40       @ddt.data(*case)
41       def test_api_01(self, data):
42         #先复制excel数据到Case文件夹下面
43         res = base_api.send_requests(self.s, data)
44         base_api.wirte_result(res, filename=reportxlsx)
45         #检查点 checkpoint
46         check = data["checkpoint"]
47         print("检查点->：%s"%check)
48         #返回结果
49         res_text = res["text"]
50         print("返回实际结果->：%s"%res_text)
51         #将接口返回的结果的键和值取到，通过键是否存在，再利用返回的值去和预期结果做断言
52         for m,n in res_text.items():
53              if m==‘data‘ and m==‘msg‘:
54                   self.assertTrue(res_text[‘data‘]!=None)
55                   self.assertTrue(res_text[‘msg‘] == check)
56              elif ‘data‘ not in m:
57                     self.assertTrue(res_text[‘msg‘]==check)