1、思路：

yamlpy即为yaml文件+pytest单元测试框架的缩写，

可以看作是一个脚手架工具，

可以快速生成项目的各个目录与文件，

只需维护一份或者多份yaml文件即可，

不需要大量写代码。

与yamlapi接口测试框架对比，

整体结构仍然保持不变，

yaml文件格式仍然保持不变，

可以通用，

抛弃了python自带的unittest单元测试框架、ddt数据驱动第三方库、BeautifulReport测试报告第三方库，

修改了测试类文件，

传参方式由ddt的@ddt.ddt、@ddt.file_data()改为pytest的@pytest.mark.parametrize()，

删掉了tool工具包里面的beautiful_report_run.py文件，

其他文件保持不变，

新增了pytest-assume多重断言插件。

（yamlapi接口测试框架也支持双重断言）

2、安装：

pip install yamlpy

请访问

https://pypi.org/project/yamlpy/

3、文件举例：

README.md文件：

 # yamlpy

 接口测试框架  

 # 一、思路

 1、采用requests+PyMySQL+demjson+loguru+PyYAML+ruamel.yaml+pytest+pytest-html+allure-pytest+pytest-assume+pytest-rerunfailures+pytest-sugar+pytest-timeout

 2、requests是发起HTTP请求的第三方库

 3、PyMySQL是连接MySQL的第三方库

 4、demjson是解析json的第三方库

 5、loguru是记录日志的第三方库

 6、PyYAML与ruamel.yaml是读写yaml文件的第三方库

 7、pytest是单元测试的第三方库

 8、pytest-html是生成html测试报告的插件

 9、allure-pytest是生成allure测试报告的插件

 10、pytest-assume是多重断言的插件

 11、pytest-rerunfailures是失败重跑的插件

 12、pytest-sugar是显示进度的插件

 13、pytest-timeout是设置超时时间的插件  

 # 二、目录结构

 1、case是测试用例包

 2、log是日志目录

 3、report是测试报告的目录

 4、resource是yaml文件的目录

 5、setting是工程的配置文件包

 6、tool是常用方法的封装包         

 # 三、yaml文件说明

 1、字段（命名和格式不可修改，顺序可以修改）

 case_name: 用例名称

 mysql: MySQL语句，-列表格式，顺序不可修改

 第一行：mysql[0]

 第二行：mysql[1]

 第三行：mysql[2]

 第一行为增删改语句，第二行为查语句，第三行为查语句（数据库双重断言）

 第一行是发起请求之前的动作，没有返回结果

 第二行是发起请求之前的动作，有返回结果，是为了动态传参

 第三行是发起请求之后的动作，有返回结果，但是不可用于动态传参，是为了断言实际的响应结果

 当不需要增删改查和双重断言时，三行都为空

 当只需要增删改时，第一行为增删改语句，第二行为空，第三行为空

 当只需要查时，第一行为空，第二行为查语句，第三行为空

 当只需要双重断言时，第一行为空，第二行为空，第三行为查语句

 request_mode: 请求方式

 api: 接口路径

 data: 请求体，缩进字典格式或者json格式

 headers: 请求头，缩进字典格式或者json格式

 query_string: 请求参数，缩进字典格式或者json格式

 expected_code: 预期的响应代码

 expected_result: 预期的响应结果，-列表格式、缩进字典格式或者json格式

 regular: 正则，缩进字典格式

 >>variable:变量名，-列表格式

 >>expression:表达式，-列表格式  

 2、参数化

 正则表达式提取的结果用${变量名}匹配，一条用例里面可以有多个

 MySQL查询语句返回的结果，即第二行mysql[1]返回的结果，用{__SQL索引}匹配

 即{__SQL0}、{__SQL1}、{__SQL2}、{__SQL3}。。。。。。一条用例里面可以有多个

 随机数字用{__RN位数}，一条用例里面可以有多个

 随机英文字母用{__RL位数}，一条用例里面可以有多个

 以上4种类型在一条用例里面可以混合使用

 ${变量名}的作用域是全局的，其它3种的作用域仅限该条用例  

 # 四、运行

 在工程的根目录下执行命令

 pytest+--cmd=环境缩写

 pytest --cmd=dev

 pytest --cmd=test

 pytest --cmd=pre

 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查询结果与实际的响应结果是被包含关系

            logger.info("##########用例分隔符##########\n")

            # 双重断言

        else:

            logger.error("{}>>>执行失败！！！", case_name)

            logger.error("预期的响应代码与实际的响应代码不相等：{}!={}", expected_code, actual_code)

            assume(expected_code == actual_code)

            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文件

最新文章

  1. 项目实现不同环境不同配置文件-maven profile
  2. Shell入门教程：Shell函数的返回值
  3. android 判断是否设置了锁屏密码
  4. Linux的phpstudy mysql登录
  5. Android笔记——我的Android课的开始
  6. Hosts文件的使用
  7. ioc开发学习 --简易计时器 （基于iPhone5屏幕尺寸开发）
  8. bzoj 4827: [Hnoi2017]礼物 [fft]
  9. PyCharm安装Pygame方法
  10. 【嵌入式开发】 Bootloader 详解 ( 代码环境 | ARM 启动流程 | uboot 工作流程 | 架构设计)
  11. Mybatis插件原理分析（二）
  12. Mysql5.6二进制包安装方法
  13. AspNetCore 基于流下载文件与示例代码
  14. Google Analytics电子商务篇(Universal版)
  15. Spring cloud Eureka错误锦集（二）
  16. 数据导入Excel时，出现ole error 800AC472这个错误，怎么解决。
  17. 初等数论及其应用 (第6版) (Kenneth H.Rosen 著)
  18. 奇怪吸引子---DequanLi
  19. code vs 1013 求先序排列
  20. Sass::SyntaxError related to active_admin/mixins

热门文章

  1. NOI2.5 1253:Dungeon Master
  2. P2869 [USACO07DEC]美食的食草动物Gourmet Grazers
  3. Scala 学习（6）之「对象」
  4. Java容器解析系列(14) IdentityHashMap详解
  5. CGI fastCgi php-fpm PHP-CGI 辨析
  6. X-CTF(REVERSE入门) re1
  7. redis 常用命令行
  8. Redis系列(三)：Redis的持久化机制(RDB、AOF)
  9. （5千字）由浅入深讲解动态规划(JS版)-钢条切割，最大公共子序列，最短编辑距离
  10. ipwry源码