yamlapi接口測試框架

一、思路：

yamlapi支持unittest與pytest兩種運行模式，

yamlapi即爲yaml文件+api測試的縮寫，

能夠看做是一個腳手架工具，

能夠快速生成項目的各個目錄與文件，

只需維護一份或者多份yaml文件便可，

不須要大量寫代碼。

二、安裝：

https://pypi.org/

可在首頁搜索「yamlapi」，

或者直接訪問項目主頁：

https://pypi.org/project/yamlapi/

 

pip install yamlapi

# 安裝

yamlapi -h（或yamlapi --help）

# 查看參數信息

yamlapi -v（或yamlapi --v）

# 查看版本號

pip install -U yamlapi

# 安裝最新版

yamlapi --p=項目名稱

# 建立項目

# 例如在某個路徑下執行命令：yamlapi --p=demo_project

pip uninstall yamlapi

# 卸載

三、工程示例：

README.md文件：

 1 # yamlapi  
 2 接口測試框架  
 3 
 4 
 5 # 1、思路         
 6 一、採用requests+unittest+ddt+PyMySQL+BeautifulReport+demjson+loguru+PyYAML+ruamel.yaml+pytest+pytest-html+allure-pytest+pytest-rerunfailures+pytest-sugar+pytest-timeout  
 7 2、requests是發起HTTP請求的第三方庫  
 8 3、unittest是Python自帶的單元測試工具  
 9 4、ddt是數據驅動的第三方庫  
10 5、PyMySQL是鏈接MySQL的第三方庫  
11 6、BeautifulReport是生成html測試報告的第三方庫  
12 7、demjson是解析json的第三方庫  
13 8、loguru是記錄日誌的第三方庫  
14 9、PyYAML與ruamel.yaml是讀寫yaml文件的第三方庫  
15 10、pytest是單元測試的第三方庫  
16 十一、pytest-html是生成html測試報告的插件  
17 十二、allure-pytest是生成allure測試報告的插件  
18 1三、pytest-rerunfailures是失敗重跑的插件   
19 1四、pytest-sugar是顯示進度的插件  
20 1五、pytest-timeout是設置超時時間的插件  
21 
22 
23 # 2、目錄結構    
24 1、case是測試用例包              
25 2、log是日誌目錄         
26 3、report是測試報告的目錄       
27 4、resource是yaml文件的目錄      
28 5、setting是工程的配置文件包            
29 6、tool是經常使用方法的封裝包         
30 
31 
32 # 3、yaml文件說明  
33 1、字段（命名和格式不可修改，順序能夠修改）  
34 case_name: 用例名稱   
35 mysql: MySQL查詢語句   
36 request_mode: 請求方式  
37 api: 接口    
38 data: 請求體，縮進字典格式或者json格式     
39 headers: 請求頭，縮進字典格式或者json格式    
40 query_string: 請求參數，縮進字典格式或者json格式    
41 expected_code: 預期的響應代碼    
42 expected_result: 預期的響應結果，-列表格式     
43 regular: 正則，縮進字典格式  
44 >>variable:變量名，-列表格式  
45 >>expression:表達式，-列表格式  
46 
47 2、參數化  
48 正則表達式提取的結果用${變量名}表示，一條用例裏面能夠有多個    
49 MySQL返回的結果用{__SQL}表示，一條用例裏面能夠有多個   
50 隨機數字用{__RN位數}，一條用例裏面能夠有多個   
51 隨機英文字母用{__RL位數}，一條用例裏面能夠有多個  
52 以上4種類型在一條用例裏面能夠混合使用  
53 ${變量名}的做用域是全局的，其它3種的做用域僅限該條用例  

demo_test.py文件：

  1 """
  2 測試用例
  3 """
  4 
  5 import json
  6 import re
  7 import os
  8 import sys
  9 import unittest
 10 from itertools import chain
 11 from time import sleep
 12 
 13 import ddt
 14 import demjson
 15 import requests
 16 
 17 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 18 sys.path.append(BASE_DIR)
 19 
 20 from setting.project_config import *
 21 from tool.connect_mysql import query_mysql
 22 from tool.create_random import create_random_number, create_random_letters
 23 from tool.read_write_yaml import read_yaml, write_yaml
 24 from tool.beautiful_report_run import beautiful_report_run
 25 
 26 
 27 @ddt.ddt
 28 # 聲明使用ddt
 29 class DemoTest(unittest.TestCase):
 30     temporary_yaml = yaml_path + "/temporary.yaml"
 31     if os.path.isfile(temporary_yaml):
 32         # 若是臨時yaml文件存在
 33         os.remove(temporary_yaml)
 34         # 刪除之
 35     demo_one_list = read_yaml("/demo_one.yaml")
 36     demo_two_list = read_yaml("/demo_two.yaml")
 37     demo_three_list = read_yaml("/demo_three.yaml")
 38     temporary_list = demo_one_list + demo_two_list + demo_three_list
 39     temporary_yaml = yaml_path + write_yaml("/temporary.yaml", temporary_list)
 40 
 41     # 把幾個yaml文件合併爲一個臨時yaml文件
 42 
 43     @classmethod
 44     def setUpClass(cls) -> None:
 45         cls.variable_result_dict = {}
 46         # 定義一個變量名與提取的結果字典
 47         # cls.variable_result_dict與self.variable_result_dict都是本類的公共屬性
 48 
 49     @ddt.file_data(yaml_path + "/temporary.yaml")
 50     # 傳入臨時yaml文件
 51     def test_demo(self, **kwargs):
 52         """
 53         測試用例
 54         :param kwargs:
 55         :return:
 56         """
 57 
 58         kwargs = str(kwargs)
 59         if "None" in kwargs:
 60             kwargs = kwargs.replace("None", "''")
 61         kwargs = demjson.decode(kwargs)
 62         # 把值爲None的替換成''空字符串，由於None沒法拼接
 63         # demjson.decode()等價於json.loads()反序列化
 64 
 65         case_name = kwargs.get("case_name")
 66         # 用例名稱
 67         self._testMethodDoc = case_name
 68         # 測試報告裏面的用例描述
 69         mysql = kwargs.get("mysql")
 70         # mysql查詢語句
 71         request_mode = kwargs.get("request_mode")
 72         # 請求方式
 73         api = kwargs.get("api")
 74         # 接口
 75         if type(api) != str:
 76             api = str(api)
 77         payload = kwargs.get("data")
 78         # 請求體
 79         if type(payload) != str:
 80             payload = str(payload)
 81         headers = kwargs.get("headers")
 82         # 請求頭
 83         if type(headers) != str:
 84             headers = str(headers)
 85         query_string = kwargs.get("query_string")
 86         # 請求參數
 87         if type(query_string) != str:
 88             query_string = str(query_string)
 89         expected_code = kwargs.get("expected_code")
 90         # 預期的響應代碼
 91         expected_result = kwargs.get("expected_result")
 92         # 預期的響應結果
 93         regular = kwargs.get("regular")
 94         # 正則
 95 
 96         logger.info("{}>>>開始執行", case_name)
 97 
 98         if environment == "prd" and mysql != "":
 99             self.skipTest("生產環境跳過此用例，請忽略")
100         # 生產環境不能鏈接MySQL數據庫，所以跳過，此行後面的都不會執行
101 
102         requests_list = [api, payload, headers, query_string]
103         # 請求數據列表
104 
105         for index, value in enumerate(requests_list):
106             # for循環修改requests_list的值
107 
108             if self.variable_result_dict:
109                 # 若是變量名與提取的結果字典不爲空
110                 if "$" in value:
111                     for key, value_2 in self.variable_result_dict.items():
112                         value = value.replace("{" + key + "}", value_2)
113                         # replace(old, new)把字符串中的舊字符串替換成正則表達式提取的值
114                     value = re.sub("\\$", "", value)
115                     # re.sub(old, new, 源字符串)默認所有替換
116                     # 若是遇到帶有轉義的字符被看成特殊字符時，使用雙反斜槓\\來轉義，或者在引號前面加r
117             else:
118                 pass
119 
120             if mysql:
121                 # 若是mysql查詢語句不爲空
122                 if "$" in mysql:
123                     # 有些場景下MySQL查詢語句也須要參數化
124                     for key, value_2 in self.variable_result_dict.items():
125                         mysql = mysql.replace("{" + key + "}", value_2)
126                     mysql = re.sub("\\$", "", mysql)
127                 mysql_result_tuple = query_mysql(mysql)
128                 # mysql查詢結果元祖
129                 mysql_result_list = list(chain.from_iterable(mysql_result_tuple))
130                 # 把二維元祖轉換爲一維列表
131                 if "__SQL" in value:
132                     for i in mysql_result_list:
133                         if type(i) != str:
134                             i = str(i)
135                         value = value.replace("{__SQL}", i, 1)
136                         # replace(old, new, 替換次數)把字符串中的{__SQL}替換成mysql查詢返回的值
137             else:
138                 pass
139 
140             if "__RN" in value:
141                 digit_list = re.findall("{__RN(.+?)}", value)
142                 # 獲取位數列表
143                 for j in digit_list:
144                     random_number = create_random_number(int(j))
145                     # 調用生成隨機數字的方法
146                     value = value.replace("{__RN" + j + "}", random_number)
147 
148             if "__RL" in value:
149                 digit_list = re.findall("{__RL(.+?)}", value)
150                 # 獲取位數列表
151                 for i in digit_list:
152                     random_letters = create_random_letters(int(i))
153                     # 調用生成隨機字母的方法
154                     value = value.replace("{__RL" + i + "}", random_letters)
155 
156             requests_list[index] = value
157 
158         api = requests_list[0]
159         payload = requests_list[1]
160         headers = requests_list[2]
161         query_string = requests_list[3]
162 
163         if payload != "":
164             payload = demjson.decode(payload)
165         if headers != "":
166             headers = demjson.decode(headers)
167         if query_string != "":
168             query_string = demjson.decode(query_string)
169 
170         url = service_domain + api
171         # 拼接完整地址
172 
173         logger.info("請求方式爲：{}", request_mode)
174         logger.info("地址爲：{}", url)
175         logger.info("請求體爲：{}", payload)
176         logger.info("請求頭爲：{}", headers)
177         logger.info("請求參數爲：{}", query_string)
178 
179         logger.info("預期的響應代碼爲：{}", expected_code)
180         logger.info("預期的響應結果爲：{}", expected_result)
181 
182         response = requests.request(
183             request_mode, url, data=json.dumps(payload),
184             headers=headers, params=query_string, timeout=(9, 15))
185         # 發起HTTP請求
186         # json.dumps()序列化把字典轉換成字符串，json.loads()反序列化把字符串轉換成字典
187         # data請求體爲字符串，headers請求頭與params請求參數爲字典
188 
189         actual_time = response.elapsed.total_seconds()
190         # 實際的響應時間
191         actual_code = response.status_code
192         # 實際的響應代碼
193         actual_result_text = response.text
194         # 實際的響應結果（文本格式）
195 
196         logger.info("實際的響應代碼爲：{}", actual_code)
197         logger.info("實際的響應結果爲：{}", actual_result_text)
198         logger.info("實際的響應時間爲：{}", actual_time)
199 
200         if regular:
201             # 若是正則不爲空
202             extract_list = []
203             # 定義一個提取結果列表
204             for i in regular["expression"]:
205                 regular_result = re.findall(i, actual_result_text)[0]
206                 # re.findall(正則表達式, 實際的響應結果)返回一個符合規則的list，取第1個
207                 extract_list.append(regular_result)
208                 # 把提取結果添加到提取結果列表裏面
209 
210             temporary_dict = dict(zip(regular["variable"], extract_list))
211             # 把變量列表與提取結果列表轉爲一個臨時字典
212 
213             for key, value in temporary_dict.items():
214                 self.variable_result_dict[key] = value
215             # 把臨時字典合併到變量名與提取的結果字典，已去重
216         else:
217             pass
218 
219         for key in list(self.variable_result_dict.keys()):
220             if not self.variable_result_dict[key]:
221                 del self.variable_result_dict[key]
222         # 刪除變量名與提取的結果字典中爲空的鍵值對
223 
224         actual_result_text = re.sub("{|}|\"|\\[|\\]", "", actual_result_text)
225         # 去除{、}、"、[與]
226         actual_result_list = re.split(":|,", actual_result_text)
227         # 把響應文本轉爲列表，並去除:與,
228 
229         if expected_code == actual_code:
230             if set(expected_result) <= set(actual_result_list):
231                 # 預期的響應結果與實際的響應結果是被包含關係
232                 # 判斷是不是其真子集
233                 logger.info("{}>>>執行經過", case_name)
234             else:
235                 logger.error("{}>>>執行失敗", case_name)
236             self.assertTrue(set(expected_result) <= set(actual_result_list))
237             # 布爾表達式斷言
238         else:
239             logger.error("{}>>>請求失敗，請檢查域名、路徑與請求參數是否正確！", url)
240             logger.error("{}>>>執行失敗", case_name)
241             self.assertTrue(set(expected_result) <= set(actual_result_list))
242 
243         logger.info("##########用例分隔符##########\n")
244         # sleep(3)
245         # 等待時間爲3秒，也能夠調整爲其餘值
246 
247 
248 if __name__ == '__main__':
249     beautiful_report_run(DemoTest)
250     # 調用BeautifulReport運行方式

project_config.py文件：

 1 """
 2 整個工程的配置文件
 3 """
 4 
 5 import os
 6 import sys
 7 import time
 8 
 9 from loguru import logger
10 
11 parameter = sys.argv[1]
12 # 從命令行獲取參數
13 
14 environment = os.getenv("measured_environment", parameter)
15 # 環境變量
16 
17 if environment == "dev":
18     service_domain = "http://www.dev.com"
19     # 開發環境
20     db_host = 'mysql.dev.com'
21     db_port = 3306
22 elif environment == "test":
23     service_domain = "http://www.test.com"
24     # 測試環境
25     db_host = 'mysql.test.com'
26     db_port = 3307
27 elif environment == "pre":
28     service_domain = "http://www.pre.com"
29     # 預生產環境
30     db_host = 'mysql.pre.com'
31     db_port = 3308
32 elif environment == "formal":
33     service_domain = "https://www.formal.com"
34     # 生產環境
35     db_host = None
36     db_port = None
37 
38 db_user = 'root'
39 db_password = '123456'
40 db_database = ''
41 # MySQL數據庫配置
42 
43 
44 current_path = os.path.dirname(os.path.dirname(__file__))
45 # 獲取當前目錄的父目錄的絕對路徑
46 # 也就是整個工程的根目錄
47 case_path = os.path.join(current_path, "case")
48 # 測試用例的目錄
49 yaml_path = os.path.join(current_path, "resource")
50 # yaml文件的目錄
51 today = time.strftime("%Y-%m-%d", time.localtime())
52 # 年月日
53 
54 report_path = os.path.join(current_path, "report")
55 # 測試報告的目錄
56 if os.path.exists(report_path):
57     pass
58 else:
59     os.mkdir(report_path, mode=0o777)
60 
61 log_path = os.path.join(current_path, "log")
62 # 日誌的目錄
63 if os.path.exists(log_path):
64     pass
65 else:
66     os.mkdir(log_path, mode=0o777)
67 
68 logging_file = os.path.join(log_path, "log{}.log".format(today))
69 
70 logger.add(
71     logging_file,
72     format="{time:YYYY-MM-DD HH:mm:ss}|{level}|{message}",
73     level="INFO",
74     rotation="500 MB",
75     encoding="utf-8",
76 )
77 # loguru日誌配置

四、運行：

unittest模式：

python+測試文件名+環境縮寫

python ./case/demo_test.py dev

python ./case/demo_test.py test

python ./case/demo_test.py pre

python ./case/demo_test.py formal

pytest模式：

pytest -v

完整命令爲：

pytest -v --reruns 3 --reruns-delay 3 --timeout=60 --junitxml=./report/report.xml --html=./report/report.html --self-contained-html --alluredir=./report/allure-report

已寫進pytest.ini配置文件