动态参数替换

需求：动态的去替换参数。

在替换的过程中，根据不同的参数自动的去找对应的值进行替换。

1. 设计思路

1. 用例数据中的槽位变量名要和对应的类属性名一致
2. 找出用例数据中的所有槽位名
3. 根据槽位名依次去类中获取对应的属性并替换

2. 封装动态参数替换函数

封装动态替换函数

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2021/11/11 20:51
# @Author  : shisuiyi
# @File    : data_handler.py
# @Software: win10 Tensorflow1.13.1 python3.9
# 测试数据处理模块

def replace_args_by_re(json_str, obj):
    """
    通过正则表达式动态去替换
    @param json_str:需要被替换的json字符串
    @param obj:通过数据的对象
    @return: 返回替换后的字符串
    """
    # 1.先找出字符串中的槽位名(槽位)
    args = re.findall('#(.*?)#', json_str)
    # 2.再去数据对象中获取对应名字的参数的值
    for arg in args:
        # 3.获取对应obj中对应参数名的属性值
        value = getattr(obj, arg, None)
        # 4.如果有属性则替换
        if value is not None:
            json_str = json_str.replace('#{}#'.format(arg), str(value))
    return json_str


if __name__ == '__main__':
    class Some:
        loan_id = 1
        token = 'abcdef'

    s = """{
"headers": {"X-Lemonban-Media-Type": "lemonban.v2","Authorization":"Bearer #token#"},
"json":{"loan_id":#loan_id#,"approved_or_not":true}
}"""
    res = replace_args_by_re(s, Some)
    print(res)

{
"headers": {"X-Lemonban-Media-Type": "lemonban.v2","Authorization":"Bearer abcdef"},
"json":{"loan_id":1,"approved_or_not":true}
}

3. 应用到项目中

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2021/12/2 21:28
# @Author  : shisuiyi
# @File    : test_audit.py
# @Software: win10 Tensorflow1.13.1 python3.9

import json
import unittest

from unittestreport import ddt, list_data

import setting
from common import logger, db
from common.data_handler import (
    get_data_from_excel,
    generate_no_usr_phone,
    replace_args_by_re)
from common.fixture import register, login, add_loan
from common.make_requests import send_http_request

cases = get_data_from_excel(setting.TEST_DATA_FILE, 'audit')


@ddt
class TestAudit(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        logger.info('===========项目审核接口开始测试===========')
        # 1.注册借钱用户
        mobile_phone = generate_no_usr_phone()
        pwd = '12345678'
        register(mobile_phone, pwd)

        # 2.登录借钱用户
        data = login(mobile_phone, pwd)

        # 保存投资用户的数据用来创建标,保存在类属性中
        # 要保存借钱用户的id和token
        cls.normal_member_id = data['id']
        cls.normal_token = data['token_info']['token']
        # 3.注册管理员用户
        mobile_phone = generate_no_usr_phone()
        register(mobile_phone, pwd, _type=0)
        # 4.登录管理员用户
        data = login(mobile_phone, pwd)
        # 保存管理员用户的token
        cls.token = data['token_info']['token']

    @classmethod
    def tearDownClass(cls) -> None:
        logger.info('===========项目审核接口结束测试===========')

    def setUp(self) -> None:
        """
        方法级前置
        :return:
        """
        # 创建项目
        res = add_loan(member_id=self.__class__.normal_member_id,
                       token=self.__class__.normal_token)

        # 将创建好的项目的id传递到测试用例中
        # 通过对象属性
        self.loan_id = res['id']

    @list_data(cases)
    def test_audit(self, item):
        """
        作业
        :param item:
        :return:
        """
        logger.info('>>>>>>>用例{}开始执行>>>>>>>>'.format(item['title']))
        # 1. 处理测试数据
        # 需要替换依赖参数
        item = json.dumps(item)  # 把用例数据dumps成字符串，一次替换
        item = replace_args_by_re(item, self)
        # item = item.replace('#loan_id#', str(self.loan_id))
        # item = item.replace('#token#', self.__class__.token)
        item = json.loads(item)

        # 再将request_data, expect_data loads为字典
        request_data = json.loads(item['request_data'])
        expect_data = json.loads(item['expect_data'])

        # 处理url
        if item['url'].startswith('http'):
            # 是否是全地址
            pass
        elif item['url'].startswith('/'):
            # 是否是短地址
            item['url'] = setting.PROJECT_HOST + item['url']
        else:
            # 接口名称
            item['url'] = setting.INTERFACES[item['url']]

        # 2. 测试步骤
        # 发送请求

        response = send_http_request(url=item['url'], method=item['method'], **request_data)

        # 3. 断言
        # 3.1 断言响应状态码
        try:
            self.assertEqual(item['status_code'], response.status_code)
        except AssertionError as e:
            logger.warning('用例【{}】响应状态码断言异常'.format(item['title']))
            logger.info('<<<<<<<<<用例{}测试结束<<<<<<<'.format(item['title']))
            raise e
        else:
            logger.info('用例【{}】响应状态码断言成功'.format(item['title']))
        # 3.2 断言响应数据
        if item['res_type'].lower() == 'json':
            res = response.json()
        elif item['res_type'].lower() == 'html':
            # 扩展思路
            res = response.text
        try:
            self.assertEqual(expect_data, {'code': res['code'], 'msg': res['msg']})
        except AssertionError as e:
            logger.warning('用例【{}】响应数据断言异常'.format(item['title']))
            logger.warning('用例【{}】期望结果为:{}'.format(item['title'], expect_data))
            logger.warning('用例【{}】的响应结果:{}'.format(item['title'], res))
            logger.info('<<<<<<<<<用例{}测试结束<<<<<<<'.format(item['title']))
            raise e
        else:
            logger.info('用例【{}】响应数据断言成功'.format(item['title']))

        # 3.3 数据库断言后面的任务
        if item.get('sql'):  # 返回指定键的值，如果键不在字典中返回默认值 None 或者设置的默认值。
            # 只有sql字段有sql的才需要校验数据库
            try:
                self.assertTrue(db.exist(item['sql']))
            except AssertionError as e:
                logger.warning('用例【{}】数据库断言异常，执行的sql为:{}'.format(item['title'], item['sql']))
                logger.info('<<<<<<<<<用例{}测试结束<<<<<<<'.format(item['title']))
                raise e
        logger.info('---------------用例{}测试成功---------------'.format(item['title']))


if __name__ == '__main__':
    unittest.main()