本次主要嘗試搭建介面自動化框架，基於 unittest+HTMLTestRunner

框架主要模組：

　　config: 存放配置檔案

　　lib: 封裝了一些介面前置函式：處理各種事物

　　log: 存放生成的日誌檔案

　　report: 放置生成的html測試報告

　　suite: 套件執行器

　　testcase: 存放測試用例

　　util: 封裝了一些公共函式（例如封裝了日誌模組，操作mysql函式，tool工具類等）

剩下的就看程式碼吧：

 1 '''
 2 對requests介面的二次封裝
 3     目的：
 4     1，統一介面呼叫的方法，為了後續的資料驅動的實現
 5
 
     2，讓測試用例更加整潔，更加乾淨
 6 '''
 7 import requests
 8 import json
 9 
10 from APITestUnittest.util import LogHandler
11 
12 
13 class HttpClient(object):
14     log = LogHandler.LogHandler().setLog()
15     """
16     eg: httpclient = HttpClient()
17     response = httpclient(method, url, data)
18     response = httpclient.send_request(method, url, data)
 
19     """
20 
21     def __init__(self):
22         self.session = requests.session()
23 
24     def send_request(self, method, url, params_type="form", data=None, **kwargs):
25         self.log.info("正在進行{0}請求，請求地址：{1},請求引數：{2}".format(method,url,data))
26         method = method.upper()
27         params_type = params_type.upper()
 
28 
29         # 如果data是字串，就將其轉換成字典
30         if isinstance(data, str):
31             data = json.loads(data)
32 
33         if "GET" == method:
34             response = self.session.request(method=method, url=url, params=data, **kwargs)
35         elif "POST" == method:
36             if 'FORM' == params_type:  # 傳送表單資料，使用data引數傳遞
37                 response = self.session.request(method=method, url=url, data=data, **kwargs)
38             else:  # "JSON" == params_type:傳送json資料，使用json從引數傳遞
39                 response = self.session.request(method=method, url=url, json=data, **kwargs)
40         elif "PUT" == method:
41             if 'FORM' == params_type:  # 傳送表單資料，使用data引數傳遞
42                 response = self.session.request(method=method, url=url, data=data, **kwargs)
43             else:  # "JSON" == params_type:傳送json資料，使用json從引數傳遞
44                 response = self.session.request(method=method, url=url, json=data, **kwargs)
45         elif "DELETE" == method:
46             if 'FORM' == params_type:  # 傳送表單資料，使用data引數傳遞
47                 response = self.session.request(method=method, url=url, data=data, **kwargs)
48             else:  # "JSON" == params_type:傳送json資料，使用json從引數傳遞
49                 response = self.session.request(method=method, url=url, json=data, **kwargs)
50         else:
51             raise ValueError('request method "{}" error'.format(method))
52         return response
53 
54     def __call__(self, method, url, params_type="form", data=None, **kwargs):
55         return self.send_request(method, url, params_type, data, **kwargs)
56 
57     def close_session(self):
58         self.session.close()

 1 '''
 2     連線資料庫： 封裝資料庫的操作函式
 3 '''
 4 import pymysql
 5 
 6 from APITestUnittest.util.LogHandler import LogHandler
 7 
 8 
 9 class Connect_Mysql(object):
10     conn = None
11     log = LogHandler().setLog()
12 
13 
14     def __init__(self, host, username, password, db, charset="utf8", port=3306):
15         self.host = host
16         self.username = username
17         self.password = password
18         self.charset = charset
19         self.db = db
20         self.port = port
21 
22     # 連線資料庫
23     def connect(self):
24         try:
25             self.conn = pymysql.connect(host=self.host,
26                                         port=self.port,
27                                         user=self.username,
28                                         password=self.password,
29                                         charset=self.charset,
30                                         db=self.db)
31             # 建立遊標
32             self.cursor = self.conn.cursor()
33         except Exception as e:
34             return e
35 
36     # 關閉資料庫連線
37     def close(self):
38         self.cursor.close()
39         self.conn.close()
40 
41     # 查詢一條資料
42     def get_one(self, sql, parmas=()):
43         ret = None
44         try:
45             self.connect()
46             self.cursor.execute(sql, parmas)
47             ret = self.cursor.fetchone()
48             #查詢結果為空
49             if ret is ():
50                 return None
51             self.close()
52         except Exception as e:
53             print(e)
54         return ret
55 
56     # 查詢所有記錄
57     def get_all(self, sql, parmas=()):
58         result = None
59         try:
60             self.connect()
61             self.cursor.execute(sql, parmas)
62             result = self.cursor.fetchall()
63             if result is ():
64                 return None
65             self.close()
66         except Exception as e:
67             print(e)
68         return result
69 
70     def __edit(self, sql, parmas):
71         count = 0
72         try:
73             self.connect()
74             count = self.cursor.execute(sql, parmas)
75             self.conn.commit()
76             self.close()
77         except Exception as e:
78             print(e)
79         return count
80 
81     # 插入
82     def insert(self, sql, parmas=()):
83         self.log.info(f"{sql}插入成功")
84         return self.__edit(sql,parmas)
85 
86     # 修改
87     def update(self, sql, parmas=()):
88         self.log.info(f"{sql}修改成功")
89         return self.__edit(sql, parmas)
90 
91     # 刪除
92     def delete(self, sql, parmas=()):
93         self.log.info(f"刪除語句{sql}刪除成功")
94         return self.__edit(sql, parmas)

 1 '''
 2 封裝了日誌類
 3 
 4 '''
 5 
 6 import logging
 7 from APITestUnittest.suites.RunCasesSuite import SuitRunner
 8 
 9 class LogHandler():
10     __log_name = SuitRunner.logname
11     # 建立一個logging物件，收集日誌
12     logger = logging.getLogger(__name__)
13     # 設定日誌的等級
14     logger.setLevel(level=logging.INFO)
15     """
16     日誌，輸出到檔案，輸出到控制檯
17     """
18     def setLog(self):
19         if not self.logger.handlers:
20             # 日誌存放路徑
21             filenamePath = f"../log/{self.__log_name}.log"
22             # 設定檔案處理器
23             __fhandler = logging.FileHandler(filename=filenamePath, encoding='utf-8')
24             # 設定控制檯處理器
25             __shandler = logging.StreamHandler()
26             # 設定格式化
27             # __format = logging.Formatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
28             __format = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
29             # 設定檔案處理格式化
30             __fhandler.setFormatter(__format)
31             # 設定控制檯處理格式化
32             __shandler.setFormatter(__format)
33             # 新增處理器
34             self.logger.addHandler(__fhandler)
35             self.logger.addHandler(__shandler)
36         return self.logger

 1 '''
 2 執行測試case套件，執行測試用例
 3     封裝套件執行器：
 4 '''
 5 
 6 import configparser
 7 import os
 8 import time
 9 import unittest
10 from HTMLTestRunner import HTMLTestRunner
11 
12 
13 class SuitRunner(object):
14     # 時間戳中不能有冒號
15     __t = time.strftime("%Y-%m-%d %H-%M-%S", time.localtime()).split("-")
16     __t2 = __t[2].split(" ")
17     __s = f"{__t[0]}年{__t[1]}月{__t2[0]}日{__t2[1]}時{__t[3]}分{__t[4]}秒"
18     logname = f"{__t[0]}年{__t[1]}月{__t2[0]}日"
19     __reportname = f"{__t[0]}年{__t[1]}月{__t2[0]}日"
20     runner = None
21 
22     def __init__(self):
23         self.config = configparser.ConfigParser()
24         config_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config")
25         self.config.read(config_dir+'/'+'env.ini', encoding='utf-8')
26         self.report_name = self.__reportname+'.'+'html'
27 
28 
29     def get_report(self, case_dir = "../testcase/wuye/", pattern = "Test*.py", **kwargs):
30         """
31         基於套件，執行，執行case生成測試報告
32         :param case_dir: case檔案所在路徑
33         :param pattern:  case檔案(匹配檔案：)
34         :return: None
35         """
36         discover = unittest.defaultTestLoader.discover(start_dir=case_dir, pattern=pattern)
37         # 測試報告配置：
38         report_dir = self.config.get("report", "report_dir")
39         name = self.config.get("report", "project_name")
40         filename = report_dir+self.report_name
41         title = f"{name}介面自動化測試報告"
42 
43         description = self.config.get("report", "description")
44         if not os.path.exists(report_dir):
45             os.mkdir(report_dir)
46         with open(filename, "wb") as file:
47             runner = HTMLTestRunner(stream=file, title=title, description=description)
48             runner.run(discover)
49 
50 
51 
52 # from APITestUnittest.suites import testdemo, casedemo
53 # from APITestUnittest.suites.testdemo import TestDemo
54 # # 單個用例新增進套件，進行執行
55 # suite = unittest.TestSuite()
56 # # 多個用例
57 # # suite.addTests([TestDemo('test_01'),TestDemo('test_02')])
58 # # 通過新增類，新增進套件
59 # suite.addTests(unittest.TestLoader().loadTestsFromTestCase(TestDemo))
60 # # 新增多個類，通過name 加入套件
61 # # suite.addTests(unittest.TestLoader().loadTestsFromNames(['testdemo.TestDemo', 'casedemo.CaseDemo']))
62 # # 建立執行器新增進入套件，進行執行
63 # runner = unittest.TextTestRunner(verbosity=2)
64 # # 2. 基於執行器來執行套件
65 # runner.run(suite)
66 
67 
68 if __name__ == '__main__':
69     test = SuitRunner()
70     test.get_report()