测试用例基类抽取

1. 抽取思路

1.1 将公用模块都封装到基类中便于子类直接调用

• 日志器
• 数据库查询器
• 项目配置

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2021/12/11 11:12
# @Author  : shisuiyi
# @File    : base_case.py
# @Software: win10 Tensorflow1.13.1 python3.9
import unittest
import setting
from common import logger, db


class BaseCase(unittest.TestCase):
   """
   用例基类
   """
   db = db
   logger = logger
   setting = setting

1.2 将测试方法中的每一个步骤单独封装成对象方法

• 测试数据的处理
• 测试步骤
• 断言

   #!/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
   from unittestreport import ddt, list_data
   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
   from common.basecase import BaseCase
   
   cases = get_data_from_excel(BaseCase.setting.TEST_DATA_FILE, 'audit')
   
   
   @ddt
   class TestAudit(BaseCase):
       @classmethod
       def setUpClass(cls) -> None:
           cls.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:
           cls.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:
           """
           self.logger.info('>>>>>>>用例{}开始执行>>>>>>>>'.format(item['title']))
           # 把测试数据绑定到方法属性case上,其他也要把一些变量绑定到对象的属性上
           self._case = item
           # 1. 处理测试数据
           self.process_test()
           # 2. 发送请求
           self.send_request()
           # 3. 断言
           self.assert_all()
           self.logger.info('---------------用例{}测试成功---------------'.format(self._case['title']))
   
       def process_test(self):
           """
           测试数据的处理
   
           """
           # 需要替换依赖参数
           self._case = json.dumps(self._case)  # 把用例数据dumps成字符串，一次替换
           self._case = replace_args_by_re(self._case, self)
           # item = item.replace('#loan_id#', str(self.loan_id))
           # item = item.replace('#token#', self.__class__.token)
           self._case = json.loads(self._case)
           # 再将request_data, expect_data loads为字典
           self._case['request_data'] = json.loads(self._case['request_data'])
           self._case['expect_data'] = json.loads(self._case['expect_data'])
   
           # 处理url
           if self._case['url'].startswith('http'):
               # 是否是全地址
               pass
           elif self._case['url'].startswith('/'):
               # 是否是短地址
               self._case['url'] = self.setting.PROJECT_HOST + self._case['url']
           else:
               # 接口名称
               self._case['url'] = self.setting.INTERFACES[self._case['url']]
       def send_request(self):
           """
           发送请求
           @return:
           """
           self.response = send_http_request(url=self._case['url'], method=self._case['method'],
                                             **self._case['request_data'])
       def assert_all(self):
           # 3.1 断言响应状态码
           try:
               self.assertEqual(self._case['status_code'], self.response.status_code)
           except AssertionError as e:
               self.logger.warning('用例【{}】响应状态码断言异常'.format(self._case['title']))
               self.logger.info('<<<<<<<<<用例{}测试结束<<<<<<<'.format(self._case['title']))
               raise e
           else:
               self.logger.info('用例【{}】响应状态码断言成功'.format(self._case['title']))
           # 3.2 断言响应数据
           if self._case['res_type'].lower() == 'json':
               res = self.response.json()
           elif self._case['res_type'].lower() == 'html':
               # 扩展思路
               res = self.response.text
           try:
               self.assertEqual(self._case['expect_data'], {'code': res['code'], 'msg': res['msg']})
           except AssertionError as e:
               self.logger.warning('用例【{}】响应数据断言异常'.format(self._case['title']))
               self.logger.warning('用例【{}】期望结果为:{}'.format(self._case['title'], self._case['expect_data']))
               self.logger.warning('用例【{}】的响应结果:{}'.format(self._case['title'], res))
               self.logger.info('<<<<<<<<<用例{}测试结束<<<<<<<'.format(self._case['title']))
               raise e
           else:
               self.logger.info('用例【{}】响应数据断言成功'.format(self._case['title']))
   
           # 3.3 数据库断言后面的任务
           if self._case.get('sql'):  # 返回指定键的值，如果键不在字典中返回默认值 None 或者设置的默认值。
               # 只有sql字段有sql的才需要校验数据库
               try:
                   self.assertTrue(self.db.exist(self._case['sql']))
               except AssertionError as e:
                   self.logger.warning('用例【{}】数据库断言异常，执行的sql为:{}'.format(self._case['title'], item['sql']))
                   self.logger.info('<<<<<<<<<用例{}测试结束<<<<<<<'.format(self._case['title']))
                   raise e
   
   
   
   if __name__ == '__main__':
       BaseCase.unittest.main()

1.3 按照第二步的思想，进一步将小步骤抽成更小的功能方法

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2021/12/11 11:12
# @Author  : shisuiyi
# @File    : base_case.py
# @Software: win10 Tensorflow1.13.1 python3.9
import json
import unittest

import setting
from common import logger, db
from common.data_handler import (
    replace_args_by_re,
    generate_no_usr_phone)
from common.make_requests import send_http_request


class BaseCase(unittest.TestCase):
    """
    用例基类
    """
    db = db
    logger = logger
    setting = setting
    name = 'base_case'

    @classmethod
    def setUpClass(cls) -> None:
        cls.logger.info('==========={}接口开始测试==========='.format(cls.name))
    @classmethod
    def tearDownClass(cls) -> None:
        cls.logger.info('==========={}接口结束测试==========='.format(cls.name))


    def flow(self, item):
        """
        测试流程
        """
        self.logger.info('>>>>>>>用例{}开始执行>>>>>>>>'.format(item['title']))
        # 把测试数据绑定到方法属性case上,其他也要把一些变量绑定到对象的属性上
        self._case = item
        # 1. 处理测试数据
        self.process_test()
        # 2. 发送请求
        self.send_request()
        # 3. 断言
        self.assert_all()


    def process_test(self):
        """
        测试数据的处理
        """
        # 1.1 生成测试数据
        self.generate_test_data()
        # 1.2 替换依赖参数
        self.replace_args()
        # 1.3 处理url
        self.process_url()

    def generate_test_data(self):
        """
        生成测试数据
        """
        """
           生成测试数据,不是固定流程，有不同可以复写
           :return:
           """
        self._case = json.dumps(self._case)
        if '$phone_number$' in self._case:
            phone = generate_no_usr_phone()
            self._case = self._case.replace('$phone_number$', phone)
        self._case = json.loads(self._case)

    def replace_args(self):
        """
        替换参数
        """
        self._case = json.dumps(self._case)  # 把用例数据dumps成字符串，一次替换
        self._case = replace_args_by_re(self._case, self)
        self._case = json.loads(self._case)
        # 再将request_data, expect_data loads为字典
        try:
            self._case['request_data'] = json.loads(self._case['request_data'])
            self._case['expect_data'] = json.loads(self._case['expect_data'])
        except Exception as e:
            self.logger.error('{}用例的测试数据格式不正确'.format(self._case['title']))
            raise e

    def process_url(self):
        """
        处理url
        """

        if self._case['url'].startswith('http'):
            # 是否是全地址
            pass
        elif self._case['url'].startswith('/'):
            # 是否是短地址
            self._case['url'] = self.setting.PROJECT_HOST + self._case['url']
        else:
            # 接口名称
            try:
                self._case['url'] = self.setting.INTERFACES[self._case['url']]
            except Exception as e:
                self.logger.error('接口名称：{}不存在'.format(self._case['url']))
                raise e


    def send_request(self):
        """
        发送请求
        @return:
        """
        self._response = send_http_request(url=self._case['url'], method=self._case['method'],
                                           **self._case['request_data'])


    def assert_all(self):
        try:
            # 3.1 断言响应状态码
            self.assert_status_code()
            # 3.2 断言响应数据
            self.assert_response()
            # 3.3 数据库断言后面的任务
            self.assert_sql()
        except  Exception as e:
            self.logger.error('++++++用例{}测试失败'.format(self._case['title']))
            raise e
        else:
            self.logger.info('<<<<<<<<<用例{}测试成功<<<<<<<'.format(self._case['title']))

    def assert_status_code(self):
        """
        断言响应状态码
        """
        try:
            self.assertEqual(self._case['status_code'], self._response.status_code)
        except AssertionError as e:
            self.logger.warning('用例【{}】响应状态码断言异常'.format(self._case['title']))
            raise e
        else:
            self.logger.info('用例【{}】响应状态码断言成功'.format(self._case['title']))

    def assert_response(self):
        """
        断言响应数据
        """
        if self._case['res_type'].lower() == 'json':
            res = self._response.json()
        elif self._case['res_type'].lower() == 'html':
            # 扩展思路
            res = self._response.text
        try:
            self.assertEqual(self._case['expect_data'], {'code': res['code'], 'msg': res['msg']})
        except AssertionError as e:
            self.logger.warning('用例【{}】响应数据断言异常'.format(self._case['title']))
            self.logger.warning('用例【{}】期望结果为:{}'.format(self._case['title'], self._case['expect_data']))
            self.logger.warning('用例【{}】的响应结果:{}'.format(self._case['title'], res))
            raise e
        else:
            self.logger.info('用例【{}】响应数据断言成功'.format(self._case['title']))

    def assert_sql(self):
        """
        断言数据库
        """
        if self._case.get('sql'):  # 返回指定键的值，如果键不在字典中返回默认值 None 或者设置的默认值。
            # 只有sql字段有sql的才需要校验数据库
            try:
                self.assertTrue(self.db.exist(self._case['sql']))
            except AssertionError as e:
                self.logger.warning('用例【{}】数据库断言异常，执行的sql为:{}'.format(self._case['title'], self._case['sql']))
                raise e

#!/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
from unittestreport import ddt, list_data
from common.fixture import register, login, add_loan
from common.make_requests import send_http_request
from common.data_handler import (
    get_data_from_excel,
    generate_no_usr_phone)
from common.base_case import BaseCase

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


@ddt
class TestAudit(BaseCase):
    name = '项目审核'
    @classmethod
    def setUpClass(cls) -> None:
        # 当子类重写了父类方法时，又想调用父类的同名方法时，就需要用到 super()
        super().setUpClass()
        # 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']


    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:
        """
        self.flow(item)


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