yamlpy接口測試框架

一、思路: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文件
相關文章
相關標籤/搜索