526互联

关键字 开发-10 封装引用自定义函数变量

发布时间 2023-12-07 19:46:10作者: dack_deng

前言

前面在yaml文件中引用内置函数以及自定义函数和变量时，都是在每个关键字后面进行单独得渲染，为了方便引用，于是我们单独对这块的内容进行封装。

1. 新增自定义函数和变量

在utils下新建自定义函数和变量的文件，my_builtins.py,新增了在接口中需要用到的一些变量和函数。这样，在传入接口参数时，就更加的灵活方便。

"""
定义一些内置函数，让yaml文件去加载并渲染
"""
import json
import random
import uuid
from faker import Faker
import time
from pathlib import Path
from typing import Any

password_my_builtins = "admin@password"
email1 = "123456@163.com"

def email2():
    return "54321@11.com"


def str_to_int(value):
    return str(value)

def admin_username():
    return "qq@163.com"


def current_time(f: str='%Y-%m-%d %H:%M:%S'):
    """获取当前时间 2022-12-16 22:13:00"""
    return time.strftime(f)

def rand_value(target: list):
    """从返回的 list 结果随机取值"""
    if isinstance(target, list):
        return target[random.randint(0, len(target)-1)]
    else:
        return target

def rand_str(len_start=None, len_end=None) -> str:
    """生成随机字符串， 如
    ${rand_str()} 得到32位字符串
    ${rand_str(3)} 得到3位字符串
    ${rand_str(3, 10)} 得到3-10位字符串
    """
    uuid_str = str(uuid.uuid4()).replace('-', '')
    print(len(uuid_str))
    if not len_start and not len_end:
        return uuid_str
    if not len_end:
        return uuid_str[:len_start]
    else:
        return uuid_str[:random.randint(len_start, len_end)]


def to_json(obj: Any) -> str:
    """python 转json"""
    return json.dumps(obj, ensure_ascii=False)

if __name__ == '__main__':
    import json

    data = {"name": "张三", "age": 30}
    json_str = json.dumps(data, ensure_ascii=False)
    json_str2 = json.dumps(data)
    print(json_str)  # 输出: {"name": "张三", "age": 30}
    print(json_str2)#表示：中文转成ascii，即 {"name": "\u5f20\u4e09", "age": 30}

2. jinja2渲染封装

jinja2在渲染变量的时候会如下缺点: 渲染的结果都是字符串类型，这会导致引用变量是数字类型的时候，无法区分数字和字符串类型。由于我们不是一次性把yaml的数据全部渲染。
我们执行用例的时候，是从上往下执行，前面用例执行提取的结果还需要给后面的变量，所以需要读取yaml数据后，渲染部分的数据。于是可以写个递归的方法，渲染获取的数据
utils/render_template_obj.py

from jinja2 import Template
import re
from utils.log import log
import jinja2
import yaml

# 解决 yaml 文件中日期被转成 datetime 类型
yaml.SafeLoader.yaml_implicit_resolvers = {
    k: [r for r in v if r[0] != 'tag:yaml.org,2002:timestamp'] for k, v in yaml.SafeLoader.yaml_implicit_resolvers.items()
}

def add(value, num=0):
    """jinja2过滤器 add函数"""
    return int(value) + num

def to_str(value):
    return f'"{value}"'

# 注册过滤器，${x | default "a"}
env_filter = jinja2.Environment(
    variable_start_string='${', variable_end_string='}'
)
env_filter.filters["add"] = add
env_filter.filters["str"] = to_str

def rend_template_str(template_str, *args, **kwargs):
    """渲染jinja2模板：
    1.改写应用变量语法，{{}} -> ${}
    2.函数应用语法：${fun()}
    :return 渲染之后的值"""
    # 解决函数内部参数引用变量
    def re_replace_template_str(match) -> str:
        """
        match: 匹配表达式
        匹配的值-> 渲染模板加载内部引用变量, 极少用到此情况
        eg: ${fun(${x})}
        """
        res_result = match.group()
        res_result_ = str(res_result).lstrip('${').rstrip('}')  # 拿到引用变量
        if '${' in res_result_ and '}' in res_result_ and res_result_.find('${') < res_result.find('}'):
            # 检查'${'和'}'这两个字符是否出现（即${后面跟着}）
            instance_temp = env_filter.from_string((res_result_))
            temp_render_res = instance_temp.render(*args, **kwargs)
            return '${' + temp_render_res + '}'
        else:
            return res_result

    # 正则替换
    template_str = re.sub('\$\{(.+)\}', re_replace_template_str, template_str)
    instance_template = env_filter.from_string(template_str)
    template_render_res = instance_template.render(*args, **kwargs)
    if template_str.startswith('${') and template_str.endswith('}') and template_str.count('${') == 1:
        # 只有一对${}
        template_raw_str = template_str.lstrip('${').rstrip('}')
        print(f'template_raw_str->{template_raw_str}')
        if kwargs.get(template_raw_str):  # 从全局容易context中拿值
            log.info(f'取值表达式：{template_raw_str}, 取值结果：{kwargs.get(template_raw_str)} {type(kwargs.get(template_raw_str))}')
            return kwargs.get(template_raw_str)
        if template_raw_str.startswith('str(') and template_raw_str.endswith(')'):
            log.info(f'取值表达式：{template_raw_str}, 取值结果：{template_render_res} {type(template_render_res)}')
            return str(template_render_res)
        if template_raw_str.startswith('int(') and template_raw_str.endswith(')'):
            log.info(f'取值表达式：{template_raw_str}, 取值结果：{template_render_res} {type(template_render_res)}')
            return int(template_render_res)
        if template_raw_str.startswith('to_json(') and template_raw_str.endswith(')'):
            log.info(f'取值表达式：{template_raw_str}, 取值结果：{template_render_res} {type(template_render_res)}')
            return str(template_render_res)
        try:
            result_value = yaml.safe_load(template_render_res) # 将字符串转成python数据类型
            # 默认'5'->int,如果需要int，则必须加上int()
            log.info(f'取值表达式：{template_raw_str}, 取值结果：{result_value} {type(result_value)}')
            return result_value
        except Exception as msg:
            log.info(f'取值表达式：{template_raw_str}, 取值结果：{template_render_res} {type(template_render_res)}')
            return template_render_res
    else:
        return template_render_res

def rend_template_obj(t_obj: dict, *args, **kwargs):
    """传dict对象，通过模板字符串递归查找模板字符串，转成新数据"""
    if isinstance(t_obj, dict):
        for key, value in t_obj.items():
            if isinstance(value, str):
                t_obj[key] = rend_template_str(value, *args, **kwargs)
            elif isinstance(value, dict):
                rend_template_obj(value, *args, **kwargs)
            elif isinstance(value, list):
                rend_template_array(value, *args, **kwargs)
            else:
                pass
    return t_obj

def rend_template_array(t_array, *args, **kwargs):
    """传list对象，通过模板字符串递归查找模板字符串"""
    if isinstance(t_array, list):
        new_array = []
        for item in t_array:
            if isinstance(item, str):
                new_array.append(rend_template_str(item, *args, **kwargs))
            elif isinstance(item, list):
                new_array.append(rend_template_array(item, *args, **kwargs))
            elif isinstance(item, dict):
                new_array.append(rend_template_obj(item, *args, **kwargs))
            else:
                new_array.append(item)
        return new_array
    else:
        return t_array

def rend_template_any(any_obj, *args, **kwargs):
    """渲染模板对象：str，dict，list"""
    if isinstance(any_obj, str):
        return rend_template_str(any_obj, *args, **kwargs)
    elif isinstance(any_obj, dict):
        return rend_template_obj(any_obj, *args, **kwargs)
    elif isinstance(any_obj, list):
        return rend_template_array(any_obj, *args, **kwargs)
    else:
        return any_obj


if __name__ == '__main__':
    contest = {}
    contest.update(vars(__builtins__))  # 调试的时候这里需要加vars(),引用yaml文件的时候确不需要，不知道为什么？

    def data_int(value):
        return str(value)

    result_var = globals()
    contest.update(result_var)
    re_str = "${int(data_int(22))}"
    result = rend_template_any(re_str, **contest)
    print(result)
    print(type(result))

2.1 优化run.py文件

在run.py文件中，我们导入上面编写的封装文件：from utils import render_template_obj，进行优化run.py文件的jiaja2渲染自定义变量的代码。

# 替换1：在run函数中，对下面的内容进行替换
# t1 = jinja2.Template(json.dumps(config_variables),
        #                     variable_start_string='${',
        #                     variable_end_string='}')
        # self.module_variables = json.loads(t1.render(**self.context))  # --> dict
        """⬇，替换上面的注释的内容"""
  self.module_variables = render_template_obj.rend_template_any(config_variables, **self.context)
  log.info(f'渲染之后的config中的变量：{self.module_variables}')


# 替换2：
    elif item == "request":
        # 渲染读取的request数据
        # t2 = jinja2.Template(json.dumps(value),variable_start_string='${', variable_end_string='}')
        # request_value = json.loads(t2.render(**self.context))
        """⬇，替换上面的注释的内容"""
        request_value = render_template_obj.rend_template_any(value, **self.context)
        log.info(f"发送request 请求： {request_value}")

config:
  name: yaml申明变量
  variables:
    username: "admin"
    password: "Admin@22"
    email11: ${email1}
    data_int: 222

test_login:
  name: 登录成功
  request:
    url: /api/v1/auth/login
    method: POST
    json:
      username: ${username}
      password: ${password}
  extract:
    code1: $.code
    code2: body.code
    token: $.data.token
    msg: '"message":"(.*?)"'
  validate:
    - eq: [$.code, "000000"]
    - eq: [$.message, OK]

test_login2:
  name: 登录失败
  request:
    url: /api/v1/auth/login
    method: POST
    json:
      username: ${email11}
      password: ${email2()}
      data_value: ${int(str_to_int(34))}

运行后，结果如下所示，2条case运行成功，jinia2代码封装优化成功。