一、思路:html
yamlpy即爲yaml文件+pytest單元測試框架的縮寫,python
能夠看做是一個腳手架工具,mysql
能夠快速生成項目的各個目錄與文件,正則表達式
只需維護一份或者多份yaml文件便可,sql
不須要大量寫代碼。數據庫
與yamlapi接口測試框架對比,express
總體結構仍然保持不變,json
yaml文件格式仍然保持不變,api
能夠通用,app
拋棄了python自帶的unittest單元測試框架、ddt數據驅動第三方庫、BeautifulReport測試報告第三方庫,
新增了pytest-assume多重斷言插件。
(yamlapi接口測試框架也支持雙重斷言)
二、安裝:
pip install yamlpy
請訪問
https://pypi.org/project/yamlpy/
三、文件舉例:
README.md文件:
1 # yamlpy 2 接口測試框架 3 4 5 # 1、思路 6 一、採用requests+PyMySQL+demjson+loguru+PyYAML+ruamel.yaml+pytest+pytest-html+allure-pytest+pytest-assume+pytest-rerunfailures+pytest-sugar+pytest-timeout 7 2、requests是發起HTTP請求的第三方庫 8 3、PyMySQL是鏈接MySQL的第三方庫 9 4、demjson是解析json的第三方庫 10 5、loguru是記錄日誌的第三方庫 11 6、PyYAML與ruamel.yaml是讀寫yaml文件的第三方庫 12 7、pytest是單元測試的第三方庫 13 八、pytest-html是生成html測試報告的插件 14 九、allure-pytest是生成allure測試報告的插件 15 十、pytest-assume是多重斷言的插件 16 十一、pytest-rerunfailures是失敗重跑的插件 17 十二、pytest-sugar是顯示進度的插件 18 1三、pytest-timeout是設置超時時間的插件 19 20 21 # 2、目錄結構 22 1、case是測試用例包 23 2、log是日誌目錄 24 3、report是測試報告的目錄 25 4、resource是yaml文件的目錄 26 5、setting是工程的配置文件包 27 6、tool是經常使用方法的封裝包 28 29 30 # 3、yaml文件說明 31 1、字段(命名和格式不可修改,順序能夠修改) 32 case_name: 用例名稱 33 mysql: MySQL語句,列表格式,順序不可修改 34 mysql[0] 35 mysql[1] 36 mysql[2] 37 第一行爲增刪改語句,第二行爲查語句,第三行爲查語句(數據庫雙重斷言) 38 第一行是發起請求以前的動做,沒有返回結果 39 第二行是發起請求以前的動做,有返回結果,是爲了動態傳參 40 第三行是發起請求以後的動做,有返回結果,可是不可用於動態傳參,是爲了斷言實際的響應結果 41 當不須要增刪改查和雙重斷言時,三行都爲空 42 當只須要增刪改時,第一行爲增刪改語句,第二行爲空,第三行爲空 43 當只須要查時,第一行爲空,第二行爲查語句,第三行爲空 44 當只須要雙重斷言時,第一行爲空,第二行爲空,第三行爲查語句 45 request_mode: 請求方式 46 api: 接口路徑 47 data: 請求體,縮進字典格式或者json格式 48 headers: 請求頭,縮進字典格式或者json格式 49 query_string: 請求參數,縮進字典格式或者json格式 50 expected_code: 預期的響應代碼 51 expected_result: 預期的響應結果,-列表格式、縮進字典格式或者json格式 52 regular: 正則,縮進字典格式 53 >>variable:變量名,-列表格式 54 >>expression:表達式,-列表格式 55 56 2、參數化 57 正則表達式提取的結果用${變量名}匹配,一條用例裏面能夠有多個 58 MySQL查詢語句返回的結果,即第二行mysql[1] 59 用{__SQL0}、{__SQL1}、{__SQL2}、{__SQL3}。。。。。。匹配,一條用例裏面能夠有多個 60 隨機數字用{__RN位數},一條用例裏面能夠有多個 61 隨機英文字母用{__RL位數},一條用例裏面能夠有多個 62 以上4種類型在一條用例裏面能夠混合使用 63 ${變量名}的做用域是全局的,其它3種的做用域僅限該條用例 64 65 66 # 4、運行 67 在工程的根目錄下執行命令 68 pytest+--cmd=環境縮寫 69 pytest --cmd=dev 70 pytest --cmd=test 71 pytest --cmd=pre 72 pytest --cmd=formal
demo_test.py文件:
""" 測試用例 """ import json import re from itertools import chain from time import sleep import allure import demjson import pytest import requests from pytest_assume.plugin import assume from setting.project_config import * from tool.connect_mysql import ConnectMySQL from tool.read_write_yaml import merge_yaml from tool.function_assistant import function_dollar, function_rn, function_rl, function_sql @allure.feature(test_scenario) class DemoTest(object): temporary_list = merge_yaml() # 調用合併全部yaml文件的方法 @classmethod def setup_class(cls): cls.variable_result_dict = {} # 定義一個變量名與提取的結果字典 # cls.variable_result_dict與self.variable_result_dict都是本類的公共屬性 @allure.story(test_story) @allure.severity(test_case_priority[0]) @allure.testcase(test_case_address, test_case_address_title) @pytest.mark.parametrize("temporary_dict", temporary_list) # 傳入臨時列表 def test_demo(self, temporary_dict): """ 測試用例 :param temporary_dict: :return: """ global mysql_result_list_after temporary_dict = str(temporary_dict) if "None" in temporary_dict: temporary_dict = temporary_dict.replace("None", "''") temporary_dict = demjson.decode(temporary_dict) # 把值爲None的替換成''空字符串,由於None沒法拼接 # demjson.decode()等價於json.loads()反序列化 case_name = temporary_dict.get("case_name") # 用例名稱 self.test_order.__func__.__doc__ = case_name # 測試報告裏面的用例描述 mysql = temporary_dict.get("mysql") # mysql語句 request_mode = temporary_dict.get("request_mode") # 請求方式 api = temporary_dict.get("api") # 接口路徑 if type(api) != str: api = str(api) payload = temporary_dict.get("data") # 請求體 if type(payload) != str: payload = str(payload) headers = temporary_dict.get("headers") # 請求頭 if type(headers) != str: headers = str(headers) query_string = temporary_dict.get("query_string") # 請求參數 if type(query_string) != str: query_string = str(query_string) expected_code = temporary_dict.get("expected_code") # 預期的響應代碼 expected_result = temporary_dict.get("expected_result") # 預期的響應結果 if type(expected_result) != str: expected_result = str(expected_result) regular = temporary_dict.get("regular") # 正則 logger.info("{}>>>開始執行", case_name) if environment == "formal" and mysql: pytest.skip("生產環境跳過此用例,請忽略") # 生產環境不能鏈接MySQL數據庫,所以跳過 if self.variable_result_dict: # 若是變量名與提取的結果字典不爲空 if mysql: if mysql[0]: mysql[0] = function_dollar(mysql[0], self.variable_result_dict.items()) # 調用替換$的方法 if mysql[1]: mysql[1] = function_dollar(mysql[1], self.variable_result_dict.items()) if mysql[2]: mysql[2] = function_dollar(mysql[2], self.variable_result_dict.items()) if api: api = function_dollar(api, self.variable_result_dict.items()) if payload: payload = function_dollar(payload, self.variable_result_dict.items()) if headers: headers = function_dollar(headers, self.variable_result_dict.items()) if query_string: query_string = function_dollar(query_string, self.variable_result_dict.items()) if expected_result: expected_result = function_dollar(expected_result, self.variable_result_dict.items()) else: pass if mysql: db = ConnectMySQL() # 實例化一個MySQL操做對象 if mysql[0]: mysql[0] = function_rn(mysql[0]) # 調用替換RN隨機數字的方法 mysql[0] = function_rl(mysql[0]) # 調用替換RL隨機字母的方法 if "INSERT" in mysql[0]: db.insert_mysql(mysql[0]) # 調用插入mysql的方法 sleep(2) # 等待2秒鐘 if "UPDATE" in mysql[0]: db.update_mysql(mysql[0]) # 調用更新mysql的方法 sleep(2) if "DELETE" in mysql[0]: db.delete_mysql(mysql[0]) # 調用刪除mysql的方法 sleep(2) if mysql[1]: mysql[1] = function_rn(mysql[1]) # 調用替換RN隨機數字的方法 mysql[1] = function_rl(mysql[1]) # 調用替換RL隨機字母的方法 if "SELECT" in mysql[1]: mysql_result_tuple = db.query_mysql(mysql[1]) # mysql查詢結果元祖 mysql_result_list = list(chain.from_iterable(mysql_result_tuple)) # 把二維元祖轉換爲一維列表 logger.info("發起請求以前mysql查詢的結果列表爲:{}", mysql_result_list) if api: api = function_sql(api, mysql_result_list) # 調用替換MySQL查詢結果的方法 if payload: payload = function_sql(payload, mysql_result_list) if headers: headers = function_sql(headers, mysql_result_list) if query_string: query_string = function_sql(query_string, mysql_result_list) if expected_result: expected_result = function_sql(expected_result, mysql_result_list) if api: api = function_rn(api) api = function_rl(api) if payload: payload = function_rn(payload) payload = function_rl(payload) payload = demjson.decode(payload) if headers: headers = function_rn(headers) headers = function_rl(headers) headers = demjson.decode(headers) if query_string: query_string = function_rn(query_string) query_string = function_rl(query_string) query_string = demjson.decode(query_string) url = service_domain + api # 拼接完整地址 logger.info("請求方式爲:{}", request_mode) logger.info("地址爲:{}", url) logger.info("請求體爲:{}", payload) logger.info("請求頭爲:{}", headers) logger.info("請求參數爲:{}", query_string) logger.info("預期的響應代碼爲:{}", expected_code) logger.info("預期的響應結果爲:{}", expected_result) response = requests.request( request_mode, url, data=json.dumps(payload), headers=headers, params=query_string, timeout=(12, 18) ) # 發起HTTP請求 # json.dumps()序列化把字典轉換成字符串,json.loads()反序列化把字符串轉換成字典 # data請求體爲字符串,headers請求頭與params請求參數爲字典 actual_time = response.elapsed.total_seconds() # 實際的響應時間 actual_code = response.status_code # 實際的響應代碼 actual_result_text = response.text # 實際的響應結果(文本格式) if mysql: if mysql[2]: mysql[2] = function_rn(mysql[2]) mysql[2] = function_rl(mysql[2]) if "SELECT" in mysql[2]: db_after = ConnectMySQL() mysql_result_tuple_after = db_after.query_mysql(mysql[2]) mysql_result_list_after = list(chain.from_iterable(mysql_result_tuple_after)) logger.info("發起請求以後mysql查詢的結果列表爲:{}", mysql_result_list_after) logger.info("實際的響應代碼爲:{}", actual_code) logger.info("實際的響應結果爲:{}", actual_result_text) logger.info("實際的響應時間爲:{}", actual_time) if regular: # 若是正則不爲空 extract_list = [] # 定義一個提取結果列表 for i in regular["expression"]: regular_result = re.findall(i, actual_result_text)[0] # re.findall(正則表達式, 實際的響應結果)返回一個符合規則的list,取第1個 extract_list.append(regular_result) # 把提取結果添加到提取結果列表裏面 temporary_dict = dict(zip(regular["variable"], extract_list)) # 把變量列表與提取結果列表轉爲一個臨時字典 for key, value in temporary_dict.items(): self.variable_result_dict[key] = value # 把臨時字典合併到變量名與提取的結果字典,已去重 else: pass for key in list(self.variable_result_dict.keys()): if not self.variable_result_dict[key]: del self.variable_result_dict[key] # 刪除變量名與提取的結果字典中爲空的鍵值對 expected_result = re.sub("{|}|\'|\"|\\[|\\]| ", "", expected_result) actual_result_text = re.sub("{|}|\'|\"|\\[|\\]| ", "", actual_result_text) # 去除大括號{、}、單引號'、雙引號"、中括號[、]與空格 expected_result_list = re.split(":|,", expected_result) actual_result_list = re.split(":|,", actual_result_text) # 把文本轉爲列表,並去除:與, logger.info("切割以後預期的響應結果列表爲:{}", expected_result_list) logger.info("切割以後實際的響應結果列表爲:{}", actual_result_list) if expected_code == actual_code: # 若是預期的響應代碼等於實際的響應代碼 if set(expected_result_list) <= set(actual_result_list): # 判斷是不是其真子集 logger.info("{}>>>預期的響應結果與實際的響應結果斷言成功", case_name) else: logger.error("{}>>>預期的響應結果與實際的響應結果斷言失敗!!!", case_name) assume(set(expected_result_list) <= set(actual_result_list)) # 預期的響應結果與實際的響應結果是被包含關係 if mysql: if mysql[2]: if set(mysql_result_list_after) <= set(actual_result_list): # 判斷是不是其真子集 logger.info("{}>>>發起請求以後mysql查詢結果與實際的響應結果斷言成功", case_name) else: logger.error("{}>>>發起請求以後mysql查詢結果與實際的響應結果斷言失敗!!!", case_name) assume(set(mysql_result_list_after) <= set(actual_result_list)) # 發起請求以後mysql查詢結果與實際的響應結果是被包含關係 # 雙重斷言 else: logger.error("{}>>>請求失敗!!!", url) logger.error("{}>>>執行失敗!!!", case_name) assume(expected_code == actual_code) assume(set(expected_result_list) <= set(actual_result_list)) logger.info("##########用例分隔符##########\n") if __name__ == "__main__": pytest.main()
project_config.py文件:
""" 整個工程的配置文件 """ import os import sys import time from loguru import logger parameter = sys.argv[1] # 從命令行獲取參數 if "--cmd=" in parameter: parameter = parameter.replace("--cmd=", "") else: pass environment = os.getenv("measured_environment", parameter) # 環境變量 if environment == "dev": service_domain = "http://www.dev.com" # 開發環境 db_host = 'mysql.dev.com' db_port = 3306 elif environment == "test": service_domain = "http://www.test.com" # 測試環境 db_host = 'mysql.test.com' db_port = 3307 elif environment == "pre": service_domain = "http://www.pre.com" # 預生產環境 db_host = 'mysql.pre.com' db_port = 3308 elif environment == "formal": service_domain = "https://www.formal.com" # 生產環境 db_host = None db_port = None db_user = 'root' db_password = '123456' db_database = '' # MySQL數據庫配置 current_path = os.path.dirname(os.path.dirname(__file__)) # 獲取當前目錄的父目錄的絕對路徑 # 也就是整個工程的根目錄 case_path = os.path.join(current_path, "case") # 測試用例的目錄 yaml_path = os.path.join(current_path, "resource") # yaml文件的目錄 today = time.strftime("%Y-%m-%d", time.localtime()) # 年月日 report_path = os.path.join(current_path, "report") # 測試報告的目錄 if os.path.exists(report_path): pass else: os.mkdir(report_path, mode=0o777) log_path = os.path.join(current_path, "log") # 日誌的目錄 if os.path.exists(log_path): pass else: os.mkdir(log_path, mode=0o777) logging_file = os.path.join(log_path, "log{}.log".format(today)) logger.add( logging_file, format="{time:YYYY-MM-DD HH:mm:ss}|{level}|{message}", level="INFO", rotation="500 MB", encoding="utf-8", ) # loguru日誌配置 test_scenario = "測試場景:XXX接口測試" test_story = "測試故事:XXX接口測試" test_case_priority = ["blocker", "critical", "normal", "minor", "trivial"] test_case_address = "http://www.testcase.com" test_case_address_title = "XXX接口測試用例地址" # allure配置 project_name = "XXX接口自動化測試" swagger_address = "http://www.swagger.com/swagger-ui.html" test_department = "測試部門:" tester = "測試人員:" # conftest配置 first_yaml = "demo_one.yaml" # 第一個yaml文件