接口自动化框架-提升编码效率-Go-Tester

前言

在落地自动化框架的时候，因为项目业务往往会需要大量测试数据来验证，每个用例前置都会先构建测试数据。这意味着每写一个用例都要复制粘贴构建数据的业务代码，同样对接口的请求和对用例的断言判断也是如此。那么就容易造成代码的编写效率不高。所以我总结了下痛点场景以及解决方法。

痛点场景

1、每次用例前置都需要请求生成数据的两个业务接口才能构建完成。业务代码需要十几行，而且要修改请求体才能生成不同类型数据。

        noteid = str(int(time.time() * 1000)) + '_noteid'
        headers = {
            'Cookie': f'sid={sid}',
            'X-user-key': f'{user_id}'
        }

        # 上传便签主体
            body = {"noteId": noteid,
                    "remindTime": 0,
                    "remindType": 0,
                    "star": 0}
        res = requests.post(url = notebody_url, headers = headers, json = body)
        # 上传便签内容
        body = {
            "title": str(int(time.time())) + '_title',
            "summary": str(int(time.time())) + '_summary',
            "body": str(int(time.time())) + '_body',
            "localContentVersion": 1,
            "noteId": noteid,
            "bodyType": 0
        }
        res = requests.post(url = note_url, headers = headers, json = body)

2、在一个测试用例里可能会出现重复利用request来实现接口请求，造成代码冗余。

       url = host + path
        headers = {
            'Cookie': 'xxxxxxxxxxxxxxxxx',
            'X-key': 'xxxxxxxxxxx'
        }
        body = {
            "groupId": groupid
        }
        res = requests.post(url, headers = headers, json = body)
        response = res.json()
      
        get_group_url = host + get_path
        get_group_body = {'excludeInValid': False}
        res = requests.post(url = get_group_url, headers = headers, json = get_group_body)
        self.assertEqual(group_id, res.json()['noteGroups'][0]['groupId'])

3、进行接口返回体校验时，需要通过断言方法来校验每个字段的存在情况、长度、类型、精确值，便会出现大量断言代码的情况。

    def testCase01_major(self):
        url = host + path
        headers = {
            'Cookie': 'xxxxxxxxxxx'
        }
        res = requests.get(url, headers = headers)
        response = res.json()
        self.assertEqual(200, res.status_code, msg = '状态码校验失败')
        self.assertEqual(int, type(response['responseTime']))
        self.assertEqual(list, type(response['webNotes']))
        self.assertEqual(str, type(response['webNotes'][0]['noteId']))

4、进行多个必填项校验时，每个必填项校验的内容几乎一样，但需要写成多条用例，导致代码量大。

   def testCase01_must_key_check(self):
        """新增分组, 必填项groupId缺失"""
        group_id = str(int(time.time() * 1000)) + '_groupId'
        body = {
            "groupId": group_id,
            "groupName": "groupName1",
            "order": 0
        }
        body.pop(‘groupId’)
        res = self.br.post(self.url, json=body, user_id=self.userId1, sid=self.sid1)
     
   def testCase02_must_key_check(self):
        """新增分组, 必填项groupName缺失"""
        group_id = str(int(time.time() * 1000)) + '_groupId'
        body = {
            "groupId": group_id,
            "groupName": "groupName1",
            "order": 0
        }
        body.pop(‘groupName’)
        res = self.br.post(self.url, json=body, user_id=self.userId1, sid=self.sid1)

解决方法

1、对于前置步骤冗余和多个用例会复用的业务方法进行封装，如数据的构建和清理。

实现可生成不同类型和数量的数据通用方法

def create_notes(user_id, sid, num, re_time=None, group_id=None):
    """ 新建数据 """
    lst = []
    for i in range(num):
        noteid = str(int(time.time() * 1000)) + '_noteid'
        headers = {
            'Cookie': f'wps_sid={sid}',
            'X-user-key': f'{user_id}'
        }
        # 上传便签主体
        if re_time:  # 上传日历便签主体
            body = {"noteId": noteid,
                    "remindTime": int(target_timestamp * 1000),
                    "remindType": 0,
                    "star": 0}
        elif group_id:
            body = {"noteId": noteid,
                    "groupId": group_id}

        else:  # 上传首页便签主体
            body = {"noteId": noteid}
        res = requests.post(url = notebody_url, headers = headers, json = body)
        # 上传便签内容
        body = {
            "title": str(int(time.time())) + '_title',
            "summary": str(int(time.time())) + '_summary',
            "body": str(int(time.time())) + '_body',
            "localContentVersion": 1,
            "noteId": noteid,
            "bodyType": 0
        }
        res = requests.post(url = note_url, headers = headers, json = body)
        lst.append(noteid)
    return lst

使用数据构建的demo

    def testCase01_major(self):
        """删除便签 主流程"""
        step('【step】前置构建便签数据')
        note_id = create_notes(self.userId1, self.sid1, 1)

2、业务层通用请求方法的封装
• 接口请求通用的异常处理方法实现（比方说请求超时、使用协议）
• 减少通用请求格式的编码成本（实现请求内容的默认值）

class BusinessRequests:
    @staticmethod
    def post(url, json=None, headers=None, sid=None, user_id=None, **kwargs):
        if headers:
            pass
        else:
            headers = {
                'Cookie': f'wps_sid={sid}',
                'X-user-key': f'{user_id}'
            }
        info(f'send url: {url}')
        info(f'send headers: {headers}')
        info(f'send body: {json}')
        try:
            res = requests.post(url, headers=headers, json=json, timeout=10, **kwargs)
        except TimeoutError:
            error('http requests Timeout!')
            raise TimeoutError
        info(f'recv code: {res.status_code}')
        info(f'recv body: {res.text}')
        return res

使用通用请求方法的demo

  def testCase01_major(self):
      res = self.br.post(self.url, json=body, user_id=self.userId1, sid=self.sid1)

3、通用的断言方法封装
• 减少用例实现断言的成本
• 规范返回体断言内容
• 精准定位断言失败问题

  class CheckPro(unittest.TestCase):
  def check_output(self, expected, actual):
      self.assertEqual(len(expected.keys()), len(actual.keys()), msg=f'{actual.keys()} object keys len inconsistent!')
      for key, value in expected.items():
          self.assertIn(key, actual.keys(), msg=f'{key} not in actual!')
          if isinstance(value, type):
              self.assertEqual(value, type(actual[key]), msg=f'{key} type inconsistent！')
          elif isinstance(value, dict):
              self.check_output(value, actual[key])
          elif isinstance(value, list):
              self.assertEqual(len(value), len(actual[key]), f'{actual.keys()} object items len inconsistent!')
              for list_index in range(len(value)):
                  if isinstance(value[list_index], type):
                      self.assertEqual(value[list_index], type(actual[key][list_index]),
                                       msg=f'{value[list_index]} type inconsistent！')
                  elif isinstance(value[list_index], dict):
                      self.check_output(value[list_index], actual[key][list_index])
                  else:
                      self.assertEqual(value[list_index], actual[key][list_index],
                                       msg=f'{value[list_index]} value inconsistent！')
          else:
              self.assertEqual(value, actual[key], msg=f'{key} value inconsistent！')

使用通用断言方法的demo

def testCase01_major(self):
    expected = {
        'responseTime': int,
        'updateTime': int
    }
    CheckPro().check_output(expected=expected, actual=res.json())

4、实现数据驱动，将测试用例需要依赖的数据和代码结构分离开来。

通过配置读取减少维护成本实现数据驱动、参数化减少编码成本
实现和规范

1.定义环境级别的配置和测试数据的配置

2.封装yaml文件的读取方法，实现不同维度配置的读取方式

class YamlRead:
   @staticmethod
   def env_config():
       """环境变量的读取方式"""
       with open(file=f'{DIR}/config/env/{ENVIRON}/config.yml', mode='r', encoding='utf-8') as f:
           return yaml.load(f, Loader=yaml.FullLoader)

   @staticmethod
   def data_config():
       with open(file=f'{DIR}/config/data/config.yml', mode='r', encoding='utf-8') as f:
           return yaml.load(f, Loader=yaml.FullLoader)

3.并且实现了环境的切换，通过全局变量ENVIRON控制环境的执行的维度

ENVIRON = 'Offline'  # 'Online' -> 线上环境， 'Offline' -> 测试环境

4.测试用例提取环境变量和测试数据的变量，并存储到配置文件当中，再通过yaml读取方法读取到用例模块的类属性下。

createGroup:
 path: 'xxxxxxxxxxxxx'
 mustKeys:
   - groupId
   - groupName
 notMustKeys:
   - order
 bodyBase: {
   "groupId": '123',
   "groupName": "groupName1",
   "order": 0
 }
 exceptBase: {}

5.实现参数化，前置需要实现用例结构，先导入三方库parameterized，使用expand的装饰器，将需要变量的参数对象通过实参传递到装饰器中，实例方法定义一个形参接收，用例中的变量关联到该形参实现参数化。

  @parameterized.expand(mustKeys)
   def testCase01_must_key_check(self, key):
       """新增分组, 必填项缺失"""
       group_id = str(int(time.time() * 1000)) + '_groupId'
       body = {
           "groupId": group_id,
           "groupName": "groupName1",
           "order": 0
       }
       body.pop(key)
       step('【step】新增分组')
       res = self.br.post(self.url, json=body, user_id=self.userId1, sid=self.sid1)
       self.assertEqual(401, res.status_code, msg='状态码校验失败')

目录CONTENT

接口自动化框架-提升编码效率

前言

痛点场景

解决方法

评论区