一、热加载：

就是在代码执行过程当中动态的调用Python中的方法达到或得动态参数的目的

在根目录下创建debug_talk.py，这里我们写入了获得随机数的方法，后面使用值直接调用即可

import random
import time
from common.yaml_util import YamlUtil

class DebugTalk:

    #获得随机数
    def get_randon_number(self,min,max):
        return random.randint(int(min),int(max))

    #读取extract.yaml文件中的值
    def read_extract_data(self,key):
        return YamlUtil().read_yaml(key)

另外根目录创建test.py,这个是获取随机时间和本地时间的一个方法，因为我后面的接口输入数据中包含一个参数是时间

import random
import time
from common.yaml_util import YamlUtil


class Test:
    #获取随机时间
    def get_random_time(self):
        # return str(int(time.time()))[1:6]     #获取随机时间，得到的是时间戳
        return time.strftime('%H:%M:%S', time.localtime(time.time()))  #获取本地时间，并将格式转换成时分秒

    #读取extract.yaml文件中的值
    def read_extract_data(self,key):
        return YamlUtil().read_yaml(key)

接下来就是request.py文件的修改：

import jsonpath
import requests
import json
from common.yaml_util import YamlUtil
from builtins import str
import re
from debug_talk import DebugTalk
from test import  Test


class RequestUtil:

    def __init__(self,two_node,obj):
        self.base_url=YamlUtil().read_config('base',two_node)
        self.obj=obj


    #替换值的方法
    # #(替换url，params,data,json,headers)
    # #(string，int,float,list,dict)
    def replace_value(self, data):
        if data:
            # 保存数据类型
            data_type = type(data)
            # 判断数据类型转换成str
            if isinstance(data, dict) or isinstance(data, list):
                str_data = json.dumps(data)
            else:
                str_data = str(data)
            for cs in range(1, str_data.count('${') + 1):
                # 替换
                if "${" in str_data and "}" in str_data:
                    start_index = str_data.index("${")
                    end_index = str_data.index("}", start_index)
                    old_value = str_data[start_index:end_index + 1]
                    print("old_value:"+old_value)
                    #反射：通过类的对象和方法字符串调用方法
                    func_name=old_value[2:old_value.index('(')]
                    args_value1=old_value[old_value.index('(')+1:old_value.index(')')]
                    new_value=""
                    if args_value1 !="" :
                        args_value2 = args_value1.split(',')
                        new_value=getattr(self.obj,func_name)(*args_value2)
                    else:
                        new_value = getattr(self.obj, func_name)()
                    str_data = str_data.replace(old_value, str(new_value))
            # 还原数据类型
            if isinstance(data, dict) or isinstance(data, list):
                data = json.loads(str_data)
            else:
                data = data_type(str_data)
        return data


    #规范yaml测试用例
    def standard_yaml(self,caseinfo):
        caseinfo_keys= caseinfo.keys()
        #判断一级关键字是否包含：name，request，validate
        if "name" in caseinfo_keys and "request" in caseinfo_keys and "validate" in caseinfo_keys:
            #判断request下面是否包含：method、url
            request_keys=caseinfo["request"].keys()
            if "method" in  request_keys and "url" in request_keys:
                print("yaml基本架构检查通过")
                method = caseinfo['request'].pop("method") #pop() 函数用于移除列表中的一个元素，并且返回该元素的值。
                url= caseinfo['request'].pop("url")
                res = self.send_request(method,url,**caseinfo['request']) #caseinfo需要解包加**
                return_text=res.text
                return_code = res.status_code
                return_json=""
                try:
                    return_json = res.json()
                except Exception as e:
                    print("extract返回的结果不是JSON格式")

                # 提取值并写入extract.yaml文件
                if "extract" in caseinfo.keys():
                    for key, value in caseinfo["extract"].items():
                        if "(.*?)" in value or "(.+?)" in value:  # 正则表达式
                            zz_value = re.search(value, return_text)
                            if zz_value:
                                extract_value = {key: zz_value.group(1)}
                                YamlUtil().write_yaml(extract_value)
                        else:  # jsonpath
                            js_value = jsonpath.jsonpath(return_json, value)
                            if js_value:
                                extract_value = {key: js_value[0]}
                                YamlUtil().write_yaml(extract_value)
            else:
                print("在request下必须包含method,url")
        else:
            print("一级关键字必须包含name,request,validate")



    sess= requests.session()

    # 统一请求封装

    def send_request(self,method,url,**kwargs):
        method=str(method).lower()  #转换小写
        #基础路径的拼接和替换
        url= self.base_url + self.replace_value(url)
        print(url)
        #参数替换
        for key,value in kwargs.items():
            if key in ['params','data','json','headers']:
                kwargs[key]=self.replace_value(value)
            elif key == "files":
                for file_key, file_path in value.items():
                    value[file_key] = open(file_path, 'rb')
        res = RequestUtil.sess.request(method, url, **kwargs)
        print(res.text)
        return res

参数为本地时间的yaml用例文件：

-
  name: 获取城市列表
  request:
    method: post
    url: /jlcloud/simulation/${read_extract_data(data)}/ranAsPlan
    json: {"time": "${get_random_time()}"}
    headers:
      'Content-Type': 'application/json'
      'X-Token': ${read_extract_data(token)}
  validate: None

这里仅使用了随机时间，如果接口包含随机数，可类比随机时间的方法

 最后需要在用例上需要创建初始化对象，用到哪个就写哪个，我这里只用到了Test()

import requests
import json
from common.request_util import RequestUtil
from common.yaml_util import YamlUtil
from test import Test

class TestRequest:
    pass
    @pytest.mark.parametrize("caseinfo",YamlUtil().read_testcase('get_token.yaml'))
    def test_login(self,caseinfo):
        res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)

    #
    @pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('userinfo.yaml'))
    def test_userinfo(self, caseinfo):
        res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)
    #     print(res.text)
    #
    @pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('city_type.yaml'))
    def test_city(self, caseinfo):
        res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)

    @pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('simulation.yaml'))
    def test_simulation(self, caseinfo):
        res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)

    @pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('ranAsPlan.yaml'))
    def test_ranAsPlan(self, caseinfo):
        res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)

二、断言

断言一般有两种，一种是通过返回值的状态码是否等于200，一种是业务的判断，业务判断可以使用两种方式，一种是想等，一种是包含

1、状态断言：返回的状态码是否等于200

2、业务断言：相等的断言、包含的断言

在request.py中进行断言方法的封装：

import jsonpath
import requests
import json
from common.yaml_util import YamlUtil
from builtins import str
import re
from debug_talk import DebugTalk
from test import  Test


class RequestUtil:

    def __init__(self,two_node,obj):
        self.base_url=YamlUtil().read_config('base',two_node)
        self.obj=obj


    #替换值的方法
    # #(替换url，params,data,json,headers)
    # #(string，int,float,list,dict)
    def replace_value(self, data):
        if data:
            # 保存数据类型
            data_type = type(data)
            # 判断数据类型转换成str
            if isinstance(data, dict) or isinstance(data, list):
                str_data = json.dumps(data)
            else:
                str_data = str(data)
            for cs in range(1, str_data.count('${') + 1):
                # 替换
                if "${" in str_data and "}" in str_data:
                    start_index = str_data.index("${")
                    end_index = str_data.index("}", start_index)
                    old_value = str_data[start_index:end_index + 1]
                    print("old_value:"+old_value)
                    #反射：通过类的对象和方法字符串调用方法
                    func_name=old_value[2:old_value.index('(')]
                    args_value1=old_value[old_value.index('(')+1:old_value.index(')')]
                    new_value=""
                    if args_value1 !="" :
                        args_value2 = args_value1.split(',')
                        new_value=getattr(self.obj,func_name)(*args_value2)
                    else:
                        new_value = getattr(self.obj, func_name)()
                    str_data = str_data.replace(old_value, str(new_value))
            # 还原数据类型
            if isinstance(data, dict) or isinstance(data, list):
                data = json.loads(str_data)
            else:
                data = data_type(str_data)
        return data


    #规范yaml测试用例
    def standard_yaml(self,caseinfo):
        caseinfo_keys= caseinfo.keys()
        #判断一级关键字是否包含：name，request，validate
        if "name" in caseinfo_keys and "request" in caseinfo_keys and "validate" in caseinfo_keys:
            #判断request下面是否包含：method、url
            request_keys=caseinfo["request"].keys()
            if "method" in  request_keys and "url" in request_keys:
                print("yaml基本架构检查通过")
                method = caseinfo['request'].pop("method") #pop() 函数用于移除列表中的一个元素，并且返回该元素的值。
                url= caseinfo['request'].pop("url")
                res = self.send_request(method,url,**caseinfo['request']) #caseinfo需要解包加**
                return_text=res.text
                return_code = res.status_code
                return_json=""
                try:
                    return_json = res.json()
                except Exception as e:
                    print("extract返回的结果不是JSON格式")

                # 提取值并写入extract.yaml文件
                if "extract" in caseinfo.keys():
                    for key, value in caseinfo["extract"].items():
                        if "(.*?)" in value or "(.+?)" in value:  # 正则表达式
                            zz_value = re.search(value, return_text)
                            if zz_value:
                                extract_value = {key: zz_value.group(1)}
                                YamlUtil().write_yaml(extract_value)
                        else:  # jsonpath
                            js_value = jsonpath.jsonpath(return_json, value)
                            if js_value:
                                extract_value = {key: js_value[0]}
                                YamlUtil().write_yaml(extract_value)
                #断言：
                self.assert_result(caseinfo['validate'],return_json,return_code)
            else:
                print("在request下必须包含method,url")
        else:
            print("一级关键字必须包含name,request,validate")



    sess= requests.session()

    # 统一请求封装

    def send_request(self,method,url,**kwargs):
        method=str(method).lower()  #转换小写
        #基础路径的拼接和替换
        url= self.base_url + self.replace_value(url)
        print(url)
        #参数替换
        for key,value in kwargs.items():
            if key in ['params','data','json','headers']:
                kwargs[key]=self.replace_value(value)
            elif key == "files":
                for file_key, file_path in value.items():
                    value[file_key] = open(file_path, 'rb')
        res = RequestUtil.sess.request(method, url, **kwargs)
        print(res.text)
        return res
    #断言
    def assert_result(self,yq_result,sj_result,return_code):
        all_flag = 0
        for yq in yq_result:
            for key,value in yq.items():
                print(key,value)
                if key=="equals":
                    flag=self.equals_assert(value,return_code,sj_result)
                    all_flag =all_flag + flag
                elif key == 'contains':
                    flag=self.contains_assert(value,sj_result)
                    all_flag = all_flag + flag
                else:
                    print("框架暂不支持此段断言方式")
        assert all_flag==0

    # 相等断言
    def equals_assert(self,value,return_code,sj_result):
        flag=0
        for assert_key,assert_value in value.items():
            print(assert_key,assert_value)
            if assert_key=="status_code":  #状态断言
                assert_value==return_code
                if assert_value!=return_code:
                    flag=flag+1
                    print("断言失败，返回的状态码不等于%s"%assert_value)
            else:
                lists=jsonpath.jsonpath(sj_result,'$..%s'%assert_key)
                if lists:
                    if assert_value not in lists:
                        flag=flag+1
                        print("断言失败："+assert_key+"不等于"+str(assert_value))
                else:
                    flag = flag + 1
                    print("断言失败：返回的结果不存在："+assert_key)
        return flag
    # 包含断言
    def contains_assert(self,value,sj_result):
        flag=0
        if value not in str(sj_result):
            flag = flag + 1
            print("断言失败：返回的结果中不包含："+value)
        return flag

然后在yaml文件中，输入断言的内容，例如我这个登录接口断言了返回值code是否等于200，以及是否包含data

完成后，让我们run起来，结果可以看到：

来源：小丫么小二郎~