python+request+excel實現接口自動化


一、說明

  該接口自動化測試時基於python+requests庫+excel進行實現的,沒有選用python+requests+uniitest+htmltestrunner或者python+requests+pytest+htmltestrunner框架:這兩個框架的好處整個框架結構全部都封裝好了,只需要往里面添加業務邏輯即接口測試py文件即可,但是有個致命的缺陷,就是測試用例的易維護性不是很好,當接口發生變化,就要去直接修改業務邏輯的py文件,而且測試數據都是寫死在每一個函數中,或者持久化在數據庫中,需要直接修改py文件,所以就沒有才這兩個框架,只選取了其中必要的組成部分:python+requests,然后自己完成后續業務邏輯;

二、接口測試框架的拆解

  現有框架是根據HTTP請求的請求、響應的生命周期來進行規划框架的目錄結構的;

  1、HTTP請求的生命周期

  1.1、建立連接;

  1.2、接收請求;

  1.3、處理請求;

  1.4、訪問資源;

  1.5、構建響應;

  1.6、發送響應;

  1.7、記錄事務處理過程;

  2、框架目錄結構

  2.1、base包:封裝請求方法

  base包下new_test.py文件,主要用於封裝常規的請求方法;

  http請求方式中主要有8種常見的請求方式:GET,POST,HEAD,OPTIONS,PUT,DELETE,TRACE,CONNECT方法,這幾種請求方式詳細使用方式請自行百度;

  我只封裝了其中最常見的四種方式:GET,POST,DELETE,PUT方式,因為項目中只用到了這四種方式;

  new_test.py:

  1 #!/usr/bin/env python
  2 # -*- coding:utf-8
  3 import  requests,json
  4 ##原主程序:將data及headers的為空的情況考慮進去,就會報錯404
  5 class Runmethod:
  6     ###get方法不會有data,查詢參數都是跟在url后面的;需要后再進行優化;
  7     def getmain(self,url,data=None,headers=None):
  8         res = None
  9         if headers !=None:  ####正常的get請求;
 10             if data == None:
 11                 res = requests.get(url=url, headers=headers)
 12             else:
 13                 res = requests.get(url=url, data=data, headers=headers)
 14 
 15         else:  ##非常規的get請求,如本身是get請求,為了保證數據的安全性直接使用get方法,將原本卸載url中的參數全部寫在data里面;
 16             if data == None:
 17                 res = requests.get(url=url)
 18             else:
 19                 res = requests.get(url=url, data=data)
 20         return res
 21 
 22     def postmain(self,url,data=None,headers=None):
 23         res = None
 24         if headers !=None: ####正常的post請求;
 25             if data ==None:
 26                 res = requests.post(url=url,headers=headers)
 27             else:
 28                 res = requests.post(url=url, data=data, headers=headers)
 29         else: ##非常規的post請求,如本身是get請求,為了保證數據的安全性直接使用post方法,將原本卸載url中的參數全部寫在data里面;
 30             if data == None:
 31                 res = requests.post(url=url)
 32             else:
 33                 res = requests.post(url=url,data=data)
 34         return res
 35     def runmain(self,method,url,data=None,headers=None):
 36         res = None
 37         if method == "GET":
 38             res = self.getmain(url,data=None,headers=None)
 39         elif method == "POST":
 40             res = self.postmain(url,data=None,headers=None)
 41         return  res
 42 ##測試類:只考慮正常情況,不考慮空值,
 43 class Runmethod2:
 44     def  getmain(self,url,data,headers):
 45         res = requests.get(url=url, data=data, headers=headers)
 46         return res
 47     def postmain(self,url,data,headers):
 48         res = requests.post(url=url, data=data, headers=headers)
 49         return res
 50     def runmain(self,method,url,data,headers):
 51         res = None
 52         if method == "GET":
 53             res = self.getmain(url, data, headers)
 54         elif method == "POST":
 55             res = self.postmain(url, data, headers)
 56         return res
 57 
 58 ##只考慮data是否是未空,前兩種方式在調試過程中有問題;
 59 class Runmethod3:
 60     ###get請求方式
 61     def  getmain(self,url,headers = None):
 62         res = None
 63         if headers !=None:
 64             res = requests.get(url=url,headers = headers)
 65         else:
 66             res = requests.get(url=url)
 67         return res.json()
 68     ##post請求方式
 69     def postmain(self,url,data=None,headers=None):
 70         res = None
 71         if headers != None:
 72             res = requests.post(url=url,data=data, headers=headers)
 73         else:
 74             res = requests.post(url=url,data=data)
 75         return res.json()
 76     def putmain(self,url,data=None,headers =None):
 77         res = None
 78         if headers != None:
 79             res = requests.post(url=url, data=data, headers=headers)
 80         else:
 81             res = requests.post(url=url, data=data)
 82         return res.json()
 83     def deletemain(self,url,data=None,headers = None):
 84         res = None
 85         if headers != None:
 86             res = requests.delete(url=url,data = data, headers=headers)
 87         else:
 88             res = requests.delete(url=url, data=data)
 89         return res.json()
 90 
 91     def runmain(self,method,url,data=None,headers=None):
 92         res = None
 93         if method == "GET":
 94             res = self.getmain(url,headers = headers)
 95         elif method == "POST":
 96             res = self.postmain(url, data=data, headers=headers)
 97         # return json.dumps(res,indent=2,ensure_ascii=False)
 98         elif method == "PUT":
 99             res = self.putmain(url,data=data,headers=headers)
100             print(type(res))
101         elif method == "DELETE":
102             res = self.deletemain(url,data=data,headers=headers)
103         return json.dumps(res,indent=2,ensure_ascii=False)
View Code

  2.2、main包:主程序執行入口

  根據讀取到的測試用例excle表中相關字段的信息進行判斷進行相關的處理;

  2.2.1、獲取到整個測試用例excel表,取到每行數據即每一條測試用例,取到每條測試用例中的每個字段,然后根據每個字段的相關信息執行每條用例;

  2.2.2、主程序執行:判斷是否執行-判斷是否有依賴-判斷依賴類型,根據依賴類型執行不同的方法-獲取到正確的依賴數據后執行用例-獲取響應報文與與預期結果對比,回寫測試執行結果-最后檢查是否需要清除測試數據-發送測試報告給對應人員; 

#!/usr/bin/env python
# -*- coding:utf-8
###下面添加系統路徑是因為代碼部署在服務器上、在windows cmd命令行下運行報錯提示找不到models添加的;
import os,sys
base_dir = os.path.dirname(os.path.abspath(__file__))
base_dir_file = os.path.dirname(base_dir)
sys.path.append(base_dir)
sys.path.append(base_dir_file)
sys.path.append(os.path.abspath(__file__))
# print(sys.path)
print(base_dir,base_dir_file)
# import sys
# import os
#
# curPath = os.path.abspath(os.path.dirname(__file__))
# rootPath = os.path.split(curPath)[0]
# PathProject = os.path.split(rootPath)[0]
# sys.path.append(rootPath)
# sys.path.append(PathProject)
"""
# 導入絕對路徑"""
# sys.path.append("D:/untitled/autotest")
# sys.path.append("D:/untitled/autotest/utils")

print(sys.path)
from base.new_test import Runmethod,Runmethod2,Runmethod3
from data.get_data import getdata
import json
from utils.common import *
from utils.op_excel import Oper_excel
from data.data_config import global_var
from data.depend_data import depentdata
from utils.send_email import sendemail1,sendemail2
from utils.op_database import database_handler
from utils.op_file import op_file_handle
###按照此時的寫法post是通了,但是get報錯了,get返回的是html文件,而程序是用json去解析的

class runtest:
    def __init__(self):
        self.run_method = Runmethod3()
        self.data = getdata()
        self.obj_iscomtent = commonutil()
        self.obj_op_excel = Oper_excel()
        self.obj_sendemail = sendemail1()
        self.obj_sendemail2 = sendemail2()
        self.obj_file_handle =op_file_handle()

    # print("========================================>>>>這是op_excel實例化打印的路徑:%s" %self.obj_op_excel)
    ##主程序執行
    def run(self):
        result = []
        total_run = 0
        pass_count = []
        fail_count = []
        #統計接口測試用例數量
        print("===================>>>主程序開始執行")
        row_counts = self.data.get_case_Lines()
        print("================>>>>測試用例總計:%s <<<<======================" %(row_counts-1))
        ###for循環下直接return是有坑的,在第一次循環的時候就直接返回了,沒有進行后續的循環;
        for i in range(1,row_counts):
            url = self.data.get_url(i)
            mehtod = self.data.get_request_method(i)
            data = self.data.get_request_datajson(i)
            # print("=====================???",data)
            # if data != None:
            #     data = json.dumps(data)
            # data = self.data.get_request_datajson(i)
            headers = self.data.header_is_run(i)
            ###判斷是否運行
            is_run = self.data.get_is_run(i)
            expect = self.data.get_expect_data(i)
            depend_type = self.data.get_depeed_type(i)
            print("執行%s條用例" %i,url,mehtod,data,is_run)
            # print(headers)
            ###判斷接口用例是否執行
            if is_run:
                total_run+=1
                ###判斷執行接口用例是否需要依賴數據
                # depend_case_id = self.data.is_depend(i)
                depend_case_id = self.data.get_depend_id(i)
                print("=========!!!!!!!依賴id",depend_case_id)
                if depend_case_id != None:
                    ##獲取依賴的響應數據
                    # try:
                    #     self.obj_dependdata = depentdata(self.data.get_depend_id(i))
                    #     depend_response = self.obj_dependdata.get_dependentdata(i)
                    # except Exception as e:
                    #     print("========>>>執行第%s條依賴用例失敗,請檢查依賴用例\r\n %s" %(i,e))
                    ###獲取數據依賴的key
                    self.obj_dependdata = depentdata(self.data.get_depend_id(i))
                    depend_response = self.obj_dependdata.get_dependentdata(i)
                    print("============>>我正在打印依賴用例的響應報文", depend_response)
                    depend_key = self.data.get_depend_field(i)
                    ###根據依賴類型進行依賴數據處理
                    """
                    依賴數據可能會存在以下幾種方式:1、header中依賴,如token;2、請求body中的參數依賴;
                    現在寫的邏輯是直接認為是body依賴,沒有考慮header依賴,我現在座的就是header依賴
                    """
                    if depend_response != None:
                        access_type = "Bearer"
                        ###header_depend需要依賴登錄接口,獲取token
                        if depend_type == "header_depend":
                            ####header依賴數據的處理邏輯,因為我的系統header依賴為token,所以我就直接做的token依賴處理
                            token = access_type + depend_response
                            print("===============>>>這是token",token)
                            headers[depend_key] = token
                            ###將登錄獲取到的token寫入到cfg.py文件中
                            self.obj_file_handle.write_file_token_handler(token)
                        elif depend_type == "body_depend":
                            pass
                elif depend_case_id == None:
                    if depend_type == "header_depend_token":
                        # headers[depend_key] = self.obj_file_handle.read_file_token_handler()
                        headers[depend_key] = self.obj_file_handle.read_file_handler("Authorization")
                        print(headers[depend_key])
                        print("++++++++++++++header_depend_token分支的這是處理依賴后的headers:",headers)
                # res = self.run_method.runmain(mehtod,url,data,headers)
                ###判斷請求方式,根據請求方式執行請求想
                # if mehtod == "GET":
                #     res = self.run_method.runmain(mehtod,url)
                # elif mehtod == "POST":
                #     res = self.run_method.runmain(mehtod, url, json.dumps(data), headers)
                res = self.run_method.runmain(mehtod,url, data = json.dumps(data), headers = headers)
                print("==========>>>>:這是返回的報文類型:",type(res))
                if self.obj_iscomtent.iscomtent(expect,res):
                    # print("測試通過")
                    self.data.write_exact_res(i,'pass')
                    pass_count.append(i)
                    print("********************這是在清理測試數據是否執行sql語句", self.data.get_clean_testdata(i))
                    if self.data.get_clean_testdata(i) == "YES":
                        clean_sql = self.data.get_clean_sql(i)
                        print("================>>>>這是需要執行的sql語句",clean_sql,type(clean_sql))
                        ##執行sql語句
                        if clean_sql != "None":
                            try:
                                obj_database_hendler = database_handler()
                                sql_excute_res = obj_database_hendler.cleantestdata(clean_sql)
                                print("=====================>>>>這是執行的sql的結果:",sql_excute_res)
                                ###寫入清除數據sql執行結果
                                self.data.write_sql_excute_res(i, sql_excute_res)
                            except Exception as e:
                                self.data.write_sql_excute_res(i, e)
                else:
                    # print("測試失敗")
                    self.data.write_exact_res(i,res)
                    fail_count.append(i)

                result.append(res)
                # print(res.text)
                ###清除測試數據
                ###獲取是否需要清除測試數據


        # return json.dumps(res.json(),ensure_ascii=False,indent=2)
        print("========測試通過的測試用例數量:%s" %len(pass_count))
        print("========測試失敗的測試用例數量:%s" %len(fail_count))
        print("========總計執行的測試用例數量:%s" %total_run)
        print("========總計未執行測試用例數量:%s" %(row_counts-1-total_run))
        print("========總計測試用例數量:%s" %(row_counts-1))
        # print(pass_count,fail_count)
        try:
            self.obj_sendemail2.send_main(pass_count,fail_count)
        except Exception as e:
            print("========================》》》發送測試郵件失敗,請檢查網絡配置:%s" %e)


        return result


if __name__ == "__main__":
    print("===============================>>>>>>當前執行路徑:%s" %os.path.abspath(__file__))
    print("=======================?????執行shell命令:%s" %os.popen("pwd").read())
    print("=======================?????執行shell命令:%s" %os.popen("cd ..").read())
    print("=======================?????執行shell命令:%s" %os.popen("pwd").read())
    obj_run  = runtest()
    res = obj_run.run()
    # for i in res:
    #     print(i)
run_autotest.py

 

  2.3、utils包:工具包,用於存放操作類方法

  2.3.1、common.py主要是用於判斷預期結果與實際結果是否一致及根據業務需要編寫特定邏輯類;

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8
 3 ###檢查接口是否正確,並檢查響應報文是否正確
 4 class commonutil:
 5     def iscomtent(self,str_one,str_two):
 6         """
 7             判斷一個參數是否在另一個參數中
 8             str_one:查找參數
 9             str_two:被查找參數
10         """
11         flag = None
12         ##可以對str_one進行數據格式的驗證
13         if str_one in str_two:
14             flag = True
15         else:
16             flag = False
17         return flag
18 
19 if __name__ == "__main__":
20     a = "a"
21     b = "adb"
22     res  = commonutil().iscomtent(a,b)
23     print(res)
common.py

 

  2.3.2、op_database.py主要是用於數據庫操作用於檢查測試數據是否正確,以及清除測試數據;

  該類直接使用的原生sql語句

 1 ####主要用戶數據庫操作
 2 import pymysql
 3 ###清除測試數據
 4 class database_handler:
 5     ###實例化就連接數據庫並獲取游標當前位置
 6     def __init__(self):
 7         self.conn = pymysql.connect(host='*****', port=3306, user='****', passwd='*****..', db='*****')
 8         ###游標,當前位置
 9         self.cursor = self.conn.cursor()
10     ###清除測試數據
11     # def cleantestdata(self,table,feild,data):
12     #     try:
13     #         sql = "delete from %s where %s = '%s'" %(table,feild,data)
14     #             # sql = "delete from tbl_user where user_name = 'autotest002'"
15     #         print(sql)
16     #         res=self.cursor.execute(sql)
17     #         self.conn.commit()
18     #         print(res)
19     #     except Exception as e:
20     #         print(e)
21     #     self.conn.close()
22 
23     def cleantestdata(self, sql_list):
24         res =""
25         try:
26             # sql = "delete from %s where %s = '%s'" % (table, feild, data)
27             # sql = "delete from tbl_user where user_name = 'autotest002'"
28             for sql in eval(sql_list):
29                 result = self.cursor.execute(sql)
30                 print("========>>>這是SQL語句的詳細情況" ,sql)
31             self.conn.commit()
32             res = "====================>>清除測試數據成功:%s" %result
33         except Exception as e:
34             res = e
35         self.conn.close()
36         return res
37 
38 if __name__ == "__main__":
39     # sql = ["update tbl_user set `enable`= 1 where id = 585"]
40     sql = ["delete from tbl_robot where robot_id = '12345678901234567890'","delete from tbl_user_robot where robot_id = '12345678901234567890';"]
41 
42     print(type(sql))
43     obj= database_handler()
44     print(obj.cleantestdata(sql))
op_database.py

 

  2.3.3、op_excel.py主要用於操作excel表-用例表,讀取數據以及寫入測試數據

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8
 3 import xlrd
 4 from xlutils.copy import copy
 5 from  data.data_config import *
 6 class Oper_excel:
 7     def __init__(self,data_path=None):
 8         if data_path:
 9             self.data_path = data_path
10             self.data_sheet = 0
11         else:
12             self.data_path = "../data_file/auto_test_case.xls"
13             self.data_sheet = 0
14         self.data = self.get_excel_data()
15     #讀取excel表的數據
16     def get_excel_data(self):
17         # print(self.data_path,self.data_sheet)
18         book = xlrd.open_workbook(self.data_path)
19         # sheet = book.sheet_by_index(self.data_sheet)
20         sheet = book.sheet_by_index(self.data_sheet)
21         # for i in sheet.get_rows():
22         #     print(i)
23         return  sheet
24     ###獲取單元格行數
25     def get_line(self):
26         tables = self.data
27         return tables.nrows
28         # return tables.ncols
29     ###獲取某個單元格的內容
30     def get_cell_value(self, row, col):
31         return self.data.cell_value(row, col)
32     ###寫入實際結果:就是修改excel表格
33     def update_data(self,row,col,value):
34         """
35         寫入excel表數據
36         :param row:
37         :param col:
38         :param value:
39         :return:
40         """
41         book = xlrd.open_workbook(self.data_path)
42         # sheet = book.sheet_by_index(self.data_sheet)
43         pre_data = copy(book)
44         sheet_data = pre_data.get_sheet(self.data_sheet)
45         sheet_data.write(row,col,value)
46         pre_data.save(self.data_path)
47     ###根據caseid查找對應行的內容
48     def get_row_data(self,caseid):
49         row_num = self.get_row_num(caseid)
50         return self.get_row_values(row_num)
51     ###根據對應的caseid找到對應的行號
52     def get_row_num(self,caseid):
53         num = 0
54         colsdata = self.get_col_data()
55         # print(colsdata)
56         for coldata in colsdata:
57             if caseid in coldata:
58                 return num
59             num+=1
60     ##獲取某一列的內容
61     def get_col_data(self,colid=None):
62         if colid !=None:
63             cols = self.data.col_values(colid)
64         else:
65             cols = self.data.col_values(0)
66         return cols
67 
68     ##根據行號找到該行的內容
69     def get_row_values(self,row):
70         tables = self.data
71         row_data =tables.row_values(row)
72         return row_data
73 
74     #excel表寫入數據
75 # print(Oper_excel)
76 if __name__ == "__main__":
77     data = Oper_excel()
78     print(data.get_line(),type(data.get_line()))
79     for i in range(data.get_line()):
80         print(i,data.get_cell_value(i,get_case_expect()))
81     print(data.get_row_data("web_api_002"))
op_excel.py

 

  2.3.4、op_file.py主要是用戶操作文本文件,對配置文件的操作:讀取和寫入(對讀取到的數據進行處理)
####處理配置文件

class op_file_handle:
    def __init__(self,data_path=None,para=None):
        if data_path:
            self.data_path = data_path
        else:
            self.data_path = "../data/"
        if para:
            self.para = para
        else:
            self.para = "cfg.text"
    def write_file_token_handler(self,token):
        ###將配置文件讀取出來
        src_f = open(self.data_path+self.para,'r',encoding="utf-8")
        comtent = src_f.readlines()
        src_f.close()
        ###修改配置文件
        dst_f = open(self.data_path+self.para,'w',encoding="utf-8",newline="")
        for data in comtent:
            if data.startswith("Authorization"):
                # dst_f.write(re.sub("Authorization = \d","Authorization ="+token,data,0))
                data_split = data.split("=")
                data_split[1] = token
                print("========>>>這是寫入的token",data_split[1])
                dst_f.write("%s = %s\n" %(data_split[0].strip(),data_split[1].strip()))
            else:
                dst_f.write(data)
        dst_f.close()
    ###讀取token配置文件,對token的處理
    def read_file_token_handler(self):
        with open(self.data_path+self.para,'r',encoding="utf-8") as f:
            for data in f.readlines():
                if data.startswith("Authorization"):
                    data_split = data.split("=")
                    print(data_split[1].strip())
                    return data_split[1].strip()
    ###依次讀取配置文件,處理配置文件信息
    def read_file_handler(self,para):
        with open(self.data_path+self.para,'r',encoding="utf-8") as f:
            for data in f.readlines():
                if data.startswith(para):
                    data_split = data.split("=")
                    # print(data_split[0].strip(),data_split[0].strip())
                    return data_split[1].strip()
###讀取郵件要發送的附件;
class op_excel_handle:
    def __init__(self, data_path=None, para=None):
        if data_path:
            self.data_path = data_path
        else:
            self.data_path = "../data_file/"
        if para:
            self.para = para
        else:
            self.para = "auto_test_case.xls"
    ###讀取郵件需要發送的附件
    def read_file_email_handler(self):
        with open(self.data_path+self.para,'rb') as f:
            cont = f.read()
        return cont




if __name__ == "__main__":
    obj_op_file_handle= op_file_handle()
    # obj_op_file_handle.write_file_handler("測試寫入")
    obj_op_file_handle.read_file_token_handler()
op_excel.py

   2.3.5、op_json.py:主要是用於操作json文件;json文件主要是存放請求body地方,存放格式是json格式

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8
 3 import json
 4 class data_read_handle:
 5     def __init__(self,data_path=None,para=None):
 6         if data_path:
 7             self.data_path = data_path
 8         else:
 9             self.data_path = "../data_file/"
10         if para:
11             self.para = para
12         else:
13             self.para = "usermanagement.json"
14         self.data = self.json_data_read()
15     def json_data_read(self):
16         f = open(self.data_path+self.para,encoding='utf-8')
17         data = f.read()
18         data = json.loads(data)
19         # print(type(data))
20         f.close()
21         # f = open(self.data_path+self.para)
22         # data = json.load(f)
23         return data
24     ###重寫json文件
25     # def json_data_reload(self,data):
26     #     f = open(self.data_path+self.para,"w",encoding="utf-8")
27     #     json.dump(data,f,ensure_ascii=False,indent=4)
28     def taskmanagement_data_read(self):
29         pass
30     ##根據參數獲取對應的body
31     def get_data_read(self,para):
32         if para == None:
33             return None
34         return self.data[para]
35 
36 if __name__ == "__main__":
37     data_read_handler= data_read_handle()
38     print(data_read_handler.data["login"])
op_json.py

 

  2.3.6、send_mail.py:smtplib以及exchangelib庫,根據公司郵箱的協議選擇不同的庫,用於發送測試報告給相關的人員;
  1 #!/usr/bin/env python
  2 # -*- coding:utf-8
  3 """
  4 郵件發送中的坑:qq可以向qq及163郵箱發送郵件;
  5 163向qq發送郵件會被郵件服務器退回,可能原因是qq郵箱在接收郵件的時候按照某種規則進行校驗處理了的;
  6 而且一定要小心過期教程的坑;
  7 后期可對郵件發送的類進行重新整合,將兩個類合並為一個類,做兼容性的處理;
  8 """
  9 import smtplib ###發送郵件的庫
 10 from email.mime.text import MIMEText  ###郵件格式的庫
 11 import time
 12 from exchangelib import DELEGATE, Account, Credentials, Message, Mailbox, HTMLBody,FileAttachment
 13 import os
 14 """
 15 1、sendemail1:使用的是smtplib庫,可用於常規郵箱如qq、163郵箱的郵件發送;
 16 2、sendemail2:使用的是exchangelib庫,可以用來發送小眾郵箱如公司級的郵箱的郵件發送;
 17 """
 18 from utils.op_file import op_file_handle,op_excel_handle
 19 class sendemail1:
 20     def __init__(self):
 21         pass
 22     ##發送郵件
 23     global send_user,email_host,password
 24     send_user = "tengjiang@countrygarden.com.cn"
 25     # email_host = "smtp.countrygarden.com.cn"
 26     email_host = "smtp-mail.outlook.com"
 27     ###授權密碼
 28     password = "654#@!qaz1"
 29     def send_mail(self,user_list,sub,content): ###接收郵件的用戶list;主題;內容
 30         user = "我有才2號"+"<"+send_user+">"
 31         message = MIMEText(content,_subtype='plain',_charset='utf-8')
 32         message['Subject'] = sub
 33         message['From'] = user
 34         message['To'] = ";".join(user_list)
 35         ##連接郵件服務器
 36         try:
 37             server = smtplib.SMTP()
 38             server.connect(email_host)
 39             server.login(send_user,password)
 40             server.sendmail(user,user_list,message.as_string())
 41             server.close()
 42             print("發送郵件成功")
 43         except smtplib.SMTPException as e:
 44             print(e)
 45     def send_main(self,pass_list,fail_list):
 46         now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
 47         pass_num = len(pass_list)
 48         fail_num = len(fail_list)
 49         count_num = pass_num+fail_num
 50         pass_rate = "%.2f%%" %(pass_num/count_num*100)
 51         fail_rate = "%.2f%%" %(fail_num/count_num*100)
 52         # user_list = ["18502337958@163.com", "309072057@qq.com"]
 53         user_list = ["yeweichao@countrygarden.com.cn"]
 54         sub = "這是我司測試部關於線上環境接口監測情況報告"
 55         content ="%s 此次運行接口測試用例總計:%s,通過用例:%s,失敗用例:%s,通過率為:%s,失敗率為:%s" %(now,count_num,pass_num,fail_num,pass_rate,fail_rate)
 56         self.send_mail(user_list,sub,content)
 57 class sendemail2:
 58         def __init__(self):
 59             self.obj_op_file_handle = op_file_handle()
 60             self.obj_op_excel_handle = op_excel_handle()
 61 
 62         ###發送郵件
 63         def send_email(self, to, subject, body):
 64             creds = Credentials(
 65                 # username='tengjiang@countrygarden.com.cn',
 66                 # password='654#@!qaz1'
 67                 username=self.obj_op_file_handle.read_file_handler("username"),
 68                 password = self.obj_op_file_handle.read_file_handler("password")
 69             )
 70             account = Account(
 71                 # primary_smtp_address='tengjiang@countrygarden.com.cn',
 72                 primary_smtp_address = self.obj_op_file_handle.read_file_handler("sender"),
 73                 credentials=creds,
 74                 autodiscover=True,
 75                 access_type=DELEGATE
 76             )
 77             m = Message(
 78                 account=account,
 79                 subject=subject,
 80                 body=HTMLBody(body),
 81                 to_recipients=[Mailbox(email_address=to)]
 82             )
 83             ###郵件附件
 84             # with open(os.path.abspath(r"../data_file/auto_test_case.xls"), "rb") as f:
 85             # with open(r"../data_file/auto_test_case.xls", "rb") as f:
 86             cont = self.obj_op_excel_handle.read_file_email_handler()
 87             attch_file = FileAttachment(name = "auto_test_case.xls",content=cont)
 88             m.attach(attch_file)
 89             m.send()
 90         ##生成發送內容並發送郵件
 91         def send_main(self, pass_list, fail_list):
 92             now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
 93             pass_num = len(pass_list)
 94             fail_num = len(fail_list)
 95             count_num = pass_num + fail_num
 96             pass_rate = "%.2f%%" % (pass_num / count_num * 100)
 97             fail_rate = "%.2f%%" % (fail_num / count_num * 100)
 98             # user_list = ["18502337958@163.com", "309072057@qq.com"]
 99             # send_list = ["tengjiang@countrygarden.com.cn", "179984387@qq.com",
100             #              "18502337958@163.com"]
101             # "yeweichao02@countrygarden.com.cn",
102             send_list = eval(self.obj_op_file_handle.read_file_handler("recv_list"))
103             subject = "這是我司測試部關於線上環境接口監測情況報告"
104             body = "%s 此次運行接口測試用例總計:%s:\r\n通過用例:%s;\r\n失敗用例:%s;\r\n通過率為:%s;\r\n失敗率為:%s;" \
105                    % (now, count_num, pass_num, fail_num, pass_rate, fail_rate)
106             for send_to in send_list:
107                 self.send_email(send_to, subject, body)
108 
109 
110 if __name__ == "__main__":
111     sen = sendemail2()
112     sen.send_main([1,2,3,4],[8,5,7,8])
send_mail.py
  2.4、data包:對excel用例的操作類方法及配置文件cfg
  2.4.1、data_config.py:主要用於存在excel表測試用例每個字段的對應關系,以及獲取每個字段的方法; 
 1 #!/usr/bin/env python
 2 # -*- coding:utf-8
 3 from utils.op_json import data_read_handle
 4 class global_var:
 5     case_id = 0  ##用例編號
 6     case_model = 1 ##測試模塊
 7     case_pre = 2 ##請求前置條件
 8     case_urlpath = 3 ##請求URL
 9     case_method = 4 ##請求方式
10     case_run = 5 ##接口是否運行
11     case_msg = 6 ##接口備注說明
12     case_header = 7  ##請求頭數
13     case_depend_type = 8
14     case_dependid = 9 ##依賴caseID
15     case_dependdata = 10 ##依賴數據
16     case_dependfileld = 11 ##依賴數據所屬字段
17     case_dataid = 12   ##請求數據
18     case_expect = 13 ##預期結果
19     case_exact = 14  ##實際結果
20     case_clean_testdata = 15 ###  是否清除測試數據 ###由於系統做的都是邏輯刪除;
21     case_clean_sql = 16
22     case_is_sql_success = 17  ###獲取執行結果
23 ###獲取每一條接口測試用例的每個字段
24 def get_case_id():
25     return global_var.case_id
26 def get_case_model():
27     return global_var.case_model
28 def get_case_pre():
29     return global_var.case_pre
30 def get_case_urlpath():
31     return global_var.case_urlpath
32 def get_case_method():
33     return global_var.case_method
34 def get_case_run():
35     return global_var.case_run
36 def get_case_msg():
37     return global_var.case_msg
38 def get_case_header():
39     return global_var.case_header
40 def get_case_depend_type():
41     return global_var.case_depend_type
42 def get_case_dependid():
43     return global_var.case_dependid
44 def get_case_dependdata():
45     return global_var.case_dependdata
46 def get_case_dependfileld():
47     return global_var.case_dependfileld
48 def get_case_dataid():
49     return global_var.case_dataid
50 def get_case_expect():
51     return global_var.case_expect
52 def get_case_exact():
53     return global_var.case_exact
54 def get_case_headers_data():
55     # data_headers = {
56     #     "Host": "10.8.202.148:10084",
57     #     "Connection": "keep-alive",
58     #     # "Content-Length": "81",
59     #     "Accept": "*/*",
60     #     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36",
61     #     "Content-Type": "application/json",
62     #     # "Origin": "http://10.8.202.148:10000",
63     #     # "Referer": "http://10.8.202.148:10000/robot-management/login.html",
64     #     "Accept-Encoding": "gzip, deflate",
65     #     "Accept-Language": "zh-CN,zh;q=0.9",
66     #     # "Authorization": "Bearer 22db77cb-9647-448c-923f-8b3c7e8e576e"
67     # }
68     obj_op_json = data_read_handle(para="header.json")
69     print(obj_op_json.json_data_read())
70     data_headers = obj_op_json.get_data_read(para="normal_headers")
71     return data_headers
72 def get_case_clean_testdata():
73     return global_var.case_clean_testdata
74 def get_case_clean_sql():
75     return global_var.case_clean_sql
76 def get_is_sql_success():
77     return global_var.case_is_sql_success
78 if __name__ == "__main__":
79     print(get_case_headers_data())
data_config.py

   2.4.2、depend_data.py:主要用處理測試依賴數據,測試依賴數據:依賴類型、依賴ID(依賴哪一條用例)、依賴數據(匹配的模式)、依賴數據所屬字段(是哪個字段需要依賴數據);該py文件主要用於執行依賴用例,然后對依賴用例的響應報文進行解析獲取依賴數據;將依賴的內容獲取到拼接到依賴的地方;

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8
 3 from utils.op_excel import Oper_excel
 4 from data.get_data import getdata
 5 import json
 6 from base.new_test import Runmethod3
 7 from jsonpath_rw import jsonpath,parse
 8 from utils.common import commonutil
 9 class depentdata:
10     """
11     通過caseid去獲取整航數據
12     """
13     def __init__(self,caseid):
14         self.caseid = caseid
15         self.opexcel = Oper_excel()
16         self.data = getdata()
17         self.run = Runmethod3()
18 
19     ###根據caseid去獲取該caseid的整行數據
20     def get_case_line_data(self):
21         return self.opexcel.get_row_data(self.caseid)
22     ###執行依賴數據的測試用例
23     def run_dependent(self):
24         row = self.opexcel.get_row_num(self.caseid)
25         url = self.data.get_url(row)
26         mehtod = self.data.get_request_method(row)
27         data = self.data.get_request_datajson(row)
28         if data != None:
29             data = json.dumps(data)
30         headers = self.data.header_is_run(row)
31         res = self.run.runmain(mehtod, url, data, headers)
32         print("正在執行依賴用例:%s" %res)
33         return res
34     ###根據規則key提取依賴數據的響應報文中的特定的數據,然后返回
35     def get_dependentdata(self,row):
36         res = self.get_dependentdata_excute(row)
37         return res
38     def get_dependentdata_excute(self,row):
39         depend_data = self.data.get_depend_key(row)
40         print("============>>依賴用例的數據", depend_data)
41         ###依賴用例執行的響應報文
42         response_data = self.run_dependent()
43         print("============>>依賴用例的響應報文", response_data)
44         ###依賴測試的預期結果
45         expect_data = self.data.get_expect_data(row)
46         print(expect_data)
47         obj_commonutil = commonutil()
48         print("判斷預期實際結果",obj_commonutil.iscomtent(expect_data,response_data))
49         if obj_commonutil.iscomtent(expect_data,response_data) == True:
50             print("++++++++++++,判斷預期響應報文包含在實際響應報文中" )
51             print(depend_data)
52             print(response_data)
53             try:
54                 json_exe = parse(depend_data)
55                 madle = json_exe.find(json.loads(response_data)["data"])
56                 res = [math.value for math in madle][0]
57                 print(res)
58                 return res
59             except Exception as e:
60                 return e
61         else:
62             print("預期的響應報文不包含在實際報文中")
63             return None
64 
65 
66 
67 
68 
69 if __name__ == "__main__":
70     ###excel中的提取規則運行有問題,需要后續檢查
71     obj_depentdata = depentdata("web_api_002")
72     print(obj_depentdata.get_dependentdata(3))
depend_data.py

  2.4.3、get_data.py:主要是用於獲取excel用例表每一個字段的內容;
  1 #!/usr/bin/env python
  2 # -*- coding:utf-8
  3 from utils.op_excel import Oper_excel
  4 from data.data_config import *
  5 from utils.op_json import *
  6 from utils.op_file import *
  7 
  8 class getdata:
  9     def __init__(self):
 10         self.oper_excel = Oper_excel()
 11     ##去獲取excel表里面的case行數
 12     def get_case_Lines(self):
 13         return self.oper_excel.get_line()
 14     ###獲取程序是否運行
 15     def get_is_run(self,row):
 16         flag = None
 17         col = get_case_run()
 18         run_model = self.oper_excel.get_cell_value(row,col)
 19         if run_model == "YES":
 20             flag = True
 21         else:
 22             flag = False
 23         return flag
 24     ###是否攜帶header
 25     def header_is_run(self,row):
 26         col = get_case_header()
 27         header = self.oper_excel.get_cell_value(row,col)
 28         if header == "YES":
 29             return get_case_headers_data()
 30         else:
 31             return None
 32     ###獲取請求方式
 33     def get_request_method(self,row):
 34         col = get_case_method()
 35         request_method = self.oper_excel.get_cell_value(row,col)
 36         return request_method
 37     ###獲取URL
 38     def get_urlpath(self,row):
 39         col = get_case_urlpath()
 40         url = self.oper_excel.get_cell_value(row,col)
 41         return url
 42     ###獲取請求數據關鍵字
 43     def get_request_dataid(self,row):
 44         col = get_case_dataid()
 45         dataid = self.oper_excel.get_cell_value(row,col)
 46         if dataid == '':
 47             dataid = None
 48         return dataid
 49     ###通過關鍵字獲取請求數據
 50     def get_request_datajson(self,row):
 51         obj_op_json =data_read_handle()
 52         data = obj_op_json.get_data_read(self.get_request_dataid(row))
 53         # data = obj_op_json.data
 54         return data
 55     ###獲取預期結果
 56     def get_expect_data(self,row):
 57         col = get_case_expect()
 58         data = self.oper_excel.get_cell_value(row,col)
 59         if data == '':
 60             data = None
 61         return data
 62     ###寫入實際結果
 63     def write_exact_res(self,row,value):
 64         col = get_case_exact()
 65         self.oper_excel.update_data(row,col,value)
 66     ###獲取依賴數據key
 67     def get_depend_key(self,row):
 68         col = get_case_dependdata()
 69         data = self.oper_excel.get_cell_value(row,col)
 70         if data == '':
 71             data = None
 72         return data
 73     ###判斷是否有數據依賴
 74     def is_depend(self,row):
 75         col = get_case_dependdata()
 76         data = self.oper_excel.get_cell_value(row, col)
 77         if data == '':
 78             data = None
 79         return data
 80     ####判斷數據依賴類型:header_depend:header依賴;body_depend:body依賴
 81     def get_depeed_type(self,row):
 82         col = get_case_depend_type()
 83         data = self.oper_excel.get_cell_value(row,col)
 84         if data == '':
 85             data = None
 86         return data
 87     ###獲取數據依賴字段
 88     def get_depend_field(self,row):
 89         col = get_case_dependfileld()
 90         data = self.oper_excel.get_cell_value(row, col)
 91         if data == '':
 92             data = None
 93         return data
 94     ###獲取依賴ID
 95     def get_depend_id(self,row):
 96         col = get_case_dependid()
 97         data = self.oper_excel.get_cell_value(row, col)
 98         if data == '':
 99             data = None
100         return data
101     ###獲取配置文件中的url_ip並且將絕對URL拼接出來
102     def get_url(self,row):
103         obj_op_file_handle = op_file_handle()
104         url_ip = obj_op_file_handle.read_file_handler("url_ip")
105         """
106         協議版本:http/https這兒暫不做處理,后期有需要在進行處理
107         """
108         url_path = self.get_urlpath(row)
109         url = "http://"+eval(url_ip)+url_path
110         return url
111     ###獲取是否清除測試數據
112     def get_clean_testdata(self,row):
113         col  = get_case_clean_testdata()
114         data = self.oper_excel.get_cell_value(row, col)
115         if data == 'NO':
116             data = None
117         elif data == 'YES':
118             return data
119     ###獲取清除數據sql
120     def get_clean_sql(self,row):
121         col = get_case_clean_sql()
122         data = self.oper_excel.get_cell_value(row, col)
123         if data == '':
124             data = None
125         return data
126     ###寫入清除數據sql執行結果
127     def write_sql_excute_res(self,row,value):
128         col = get_is_sql_success()
129         print(col)
130         self.oper_excel.update_data(row,col,value)
131 if __name__ == "__main__":
132         obj_getdata = getdata()
133 #         print(obj_getdata.get_case_Lines())
134 #         ###根據請求id到json文件中拿到請求數據
135 #         for i in range(1,obj_getdata.get_case_Lines()):
136 #             print(obj_getdata.get_request_datajson(i))
137 #         print(obj_getdata.get_depend_key(3))
138         print(obj_getdata.get_request_datajson(3))
get_data.py

   2.4.4、cfg.text:配置文件,存放常用的變量,以及可以對特定的變量進行修改;

 1 ###服務器主機名
 2 url_ip = "****" ###127.0.0.1:8080
 3 ###token 獲取到token持久化,后續token依賴都到配置文件中取值;
 4 ###token刷新或者過期重新持久化
 5 Authorization = Bearer2e4c4b2e-93d1-45fd-9201-8d778e4f841c
 6 ###測試報告郵件登錄賬號
 7 username =****  ##111111111@qq.com
 8 ###測試報告郵件登錄密碼
 9 password = ****
10 ###測試報告郵件發送人地址
11 sender =****  ##111111111@qq.com
12 ###測試報告郵件接收人地址
13 recv_list = ["****", "****","****"]
cfg.text

  2.5、data_file包:用於存放測試用例及json文件

  2.5.1、auto_test_case.xls:存放接口測試用例;

  

接口測試用例表每個字段備注:

auto_test_case.xls:存放接口測試用例;
id:接口測試用例id;
模塊名稱:接口所屬模塊
前提條件:接口依賴的前提條件
接口地址:只放置了接口path,接口的url可以通過讀取配置文件獲取;
請求方式:http協議的常規請求方式
是否執行:通過讀取該字段可以控制用例是否執行;
接口說明:描述該接口是干嘛的;
header:是否需要在發送請求的時候傳遞headers
依賴類型:依賴的類型:屬於headers依賴還是數據body依賴;
依賴id:依賴於哪一條接口測試用例;
依賴數據:對響應報文解析的模式,讀取模式對響應報文進行解析,獲取依賴的值;
依賴數據所屬字段:是哪一個字段需要依賴數據,獲取到依賴的值之后對依賴數據進行拼接然后發起請求
請求數據:存放請求body的key,然后根據key值到指定的json文件中讀取請求body;
預期結果:存放預期結果的key,然后根據key值到指定的json文件中讀取預期結果;
實際結果:存放實際結果,在請求完成后獲取到結果然后寫入到對應的接口測試用例行中;
清除測試數據:是否需要清除測試數據,
執行sql語句:如果需要清除數據,然后調用數據庫操作即可
執行結果:將sql語句的執行結果吸入到改表格內
備注

  2.5.2、header.json:主要用於存在常規通過的headers

  

  2.5.3、usermanagement.json:用於存放請求body,json格式,可以根據不同的模塊創建不同的json文件; 

三、項目部署

  1、python3環境的部署

  python3在Linux系統上的在線部署網上到處有,今天主要記錄一下離線安裝;

  1.1、離線安裝python解釋器

  a、python官網上:https://www.python.org/downloads/source/ 上下載tar.gz壓縮包;

  b、通過smtp上傳到服務;

  c、解壓;

  d、編譯:

1 cd Python-3.7.2     # 進入python3安裝包目錄
2 ./configure --prefix=/usr/local/python3    # 將python3安裝在這個目錄
3 make && make install    # 編譯和安裝

 

  e、創建軟連接 

1 ln -s /usr/local/python3/bin/python3 /usr/bin/python3    # 創建python3軟連接
2 ln -s /usr/local/python3/bin/pip3 /usr/bin/pip3    # 創建pip3的軟連接

 

  備注:為何要創建這樣的軟連接:
  Linux系統中編譯安裝python3創建軟連接其實和windows上添加環境python3的環境變量是一樣作用的,Linux在/usr/bin/ 目錄下創建軟連接就類似於將python3添加到環境變量中;

  不信,你可以使用echo打印一下當前環境變量:echo $PATH,當前用戶的環境變量如下;

  

  1.2、離線安裝python3插件

  a、運行python程序,找到報錯原因,確定報錯原因是否是缺少插件造成的;

  b、復制報錯中的插件,進入官網:https://www.python.org/ 搜索對應的插件;或者搜索引擎中直接輸入python3+包名;

  

  c、下載成功上傳到服務器上,下載下來的插件有幾種格式:tar.gz/zip/whl

  d、安裝對應的插件,這是我離線下載的python插件,報錯提示缺什么就對應下載什么插件;

  對於不同格式的插件安裝方式有所不同;

  d.1、tar.gz、zip包直接解壓,進入目錄,對於這類插件,目錄里都有setup.py文件,直接python3 setup.py install即可安裝成功,直接將插件安裝到虛擬運行環境sites-packages目錄下;

  d.2、whl格式的插件安裝,直接 pip3 install 對應的whl文件即可,也是直接將插件安裝到虛擬運行環境sites-packages目錄下;

到目前為止只遇到這兩個方式安裝python3的插件;

  2、jenkins部署

  2.1、jenkins安裝

  到目前為止接觸到兩種方式:war包放在tomcat下運行;jar包直接運行,需要依賴java;

  2.2、jenkins構建工程;

  中間過程中的jenkins插件安裝及配置就不做說明了,網上大把資源等着你去搜索;主要詳細記錄一下jenkins工程構建過程;

  環境配置說明:因為公司已經使用堡壘機登錄,不允許直接通過終端程序登錄遠程服務器,所以jenkins,原計划是jenkins在A服務器上,接口測試項目在B服務器上,因此就直接將jenkins服務和接口自動化的項目放在同一服務器上;

  本次部署的構建工程非常簡單:

  a、源碼管理,因為代碼通過公司自搭建的git管理,配置git參數后,可以直接將配置倉庫的指定分支代碼拉到jenkins的工作目錄下;

 

 

 

 

   b、構建、直接選擇執行shell;代碼拉到jenkins的工作目錄下,沒有進行打包解壓編譯、清除文件、記錄日志等相關操作,所以直接在工作目錄下直接運行的程序;

  

 

   就這樣完成了初級的jenkins拉取遠程倉庫的指定分支的代碼,然后在服務器上運行的操作;

  3、接口自動化測試項目部署

  其實接口自動化測試項目在服務器上的部署就是前面python環境的部署已經jenkins構建工程的相關操作,哈哈!


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM