接口自动化测试框架开发 | Pytest+Allure+AIOHTTP+用例自动生成

测试开发实战技能进阶学习，文末加群！
近期准备优先做接口测试的覆盖，为此需要开发一个测试框架，经过思考，这次依然想做点儿不一样的东西。

接口测试是比较讲究效率的，测试人员会希望很快能得到结果反馈，然而接口的数量一般都很多，而且会越来越多，所以提高执行效率很有必要；
- 接口测试的用例其实也可以用来兼做简单的压力测试，而压力测试需要并发；
- 接口测试的用例有很多重复的东西，测试人员应该只需要关注接口测试的设计，这些重复劳动最好自动化来做；
- Pytest 和 Allure 太好用了，新框架要集成它们；
- 接口测试的用例应该尽量简洁，最好用 yaml，这样数据能直接映射为请求数据，写起用例来跟做填空题一样，便于向没有自动化经验的成员推广；
- 加上我对 Python 的协程很感兴趣，也学了一段时间，一直希望学以致用，所以 HTTP 请求我决定用 AIOHTTP 来实现；
  但是 pytest 是不支持事件循环的，如果想把它们结合还需要一番功夫。于是继续思考，思考的结果是其实我可以把整个事情分为两部分；

第一部分，读取 yaml 测试用例，HTTP 请求测试接口，收集测试数据。

第二部分，根据测试数据，动态生成 pytest 认可的测试用例，然后执行，生成测试报告。

这样一来，两者就能完美结合了，也完美符合我所做的设想。想法既定，接着就是实现了。

第一部分

整个过程都要求是异步非阻塞的

读取 yaml 测试用例

一份简单的用例模板我是这样设计的，这样的好处是，参数名和
aioHTTP.ClientSession().request(method,url,**kwargs)
是直接对应上的，我可以不费力气的直接传给请求方法，避免各种转换，简洁优雅，表达力又强。

        args:  - post  -       - /xxx/add  -     kwargs:  -       -  -         caseName: 新增 xxx  -         data:  -           name: ${gen_uid(10)}  -     validator:  -       -  -         json:  -           successed: True  -

异步读取文件可以使用 aiofiles 这个第三方库，yaml_load 是一个协程，可以保证主进程读取 yaml 测试用例时不被阻塞，通过await yaml_load()便能获取测试用例的数据

        async def yaml_load(dir='', file=''):  """  异步读取 yaml 文件，并转义其中的特殊值  :param file:  :return:  """  if dir:  file = os.path.join(dir, file)  async with aiofiles.open(file, 'r', encoding='utf-8', errors='ignore') as f:  data = await f.read()  data = yaml.load(data)  # 匹配函数调用形式的语法  pattern_function = re.compile(r'^\${([A-Za-z_]+\w*\(.*\))}$')  pattern_function2 = re.compile(r'^\${(.*)}$')  # 匹配取默认值的语法  pattern_function3 = re.compile(r'^\$\((.*)\)$')  def my_iter(data):  """  递归测试用例，根据不同数据类型做相应处理，将模板语法转化为正常值  :param data:  :return:  """  if isinstance(data, (list, tuple)):  for index, _data in enumerate(data):  data[index] = my_iter(_data) or _data  elif isinstance(data, dict):  for k, v in data.items():  data[k] = my_iter(v) or v  elif isinstance(data, (str, bytes)):  m = pattern_function.match(data)  if not m:  m = pattern_function2.match(data)  if m:  return eval(m.group(1))  if not m:  m = pattern_function3.match(data)  if m:  K, k = m.group(1).split(':')  return bxmat.default_values.get(K).get(k)  return data  my_iter(data)  return BXMDict(data)

可以看到，测试用例还支持一定的模板语法，如${function}、$(a:b)等，这能在很大程度上拓展测试人员用例编写的能力

HTTP 请求测试接口

HTTP 请求可以直接用 aioHTTP.ClientSession().request(method,url,**kwargs)，HTTP
也是一个协程，可以保证网络请求时不被阻塞，通过await HTTP()便可以拿到接口测试数据。

        async def HTTP(domain, *args, **kwargs):  """  HTTP 请求处理器  :param domain: 服务地址  :param args:  :param kwargs:  :return:  """  method, api = args  arguments = kwargs.get('data') or kwargs.get('params') or kwargs.get('json') or {}  # kwargs 中加入 token  kwargs.setdefault('headers', {}).update({'token': bxmat.token})  # 拼接服务地址和 api  url = ''.join([domain, api])  async with ClientSession() as session:  async with session.request(method, url, **kwargs) as response:  res = await response_handler(response)  return {  'response': res,  'url': url,  'arguments': arguments  }

收集测试数据

协程的并发真的很快，这里为了避免服务响应不过来导致熔断，可以引入asyncio.Semaphore(num)来控制并发。

        async def entrace(test_cases, loop, semaphore=None):  """  HTTP 执行入口  :param test_cases:  :param semaphore:  :return:  """  res = BXMDict()  # 在 CookieJar 的 update_cookies 方法中，如果 unsafe=False 并且访问的是 IP 地址，客户端是不会更新 cookie 信息  # 这就导致 session 不能正确处理登录态的问题  # 所以这里使用的 cookie_jar 参数使用手动生成的 CookieJar 对象，并将其 unsafe 设置为 True  async with ClientSession(loop=loop, cookie_jar=CookieJar(unsafe=True), headers={'token': bxmat.token}) as session:  await advertise_cms_login(session)  if semaphore:  async with semaphore:  for test_case in test_cases:  data = await one(session, case_name=test_case)  res.setdefault(data.pop('case_dir'), BXMList()).append(data)  else:  for test_case in test_cases:  data = await one(session, case_name=test_case)  res.setdefault(data.pop('case_dir'), BXMList()).append(data)  return res  async def one(session, case_dir='', case_name=''):  """  一份测试用例执行的全过程，包括读取 .yml 测试用例，执行 HTTP 请求，返回请求结果  所有操作都是异步非阻塞的  :param session: session 会话  :param case_dir: 用例目录  :param case_name: 用例名称  :return:  """  project_name = case_name.split(os.sep)[1]  domain = bxmat.url.get(project_name)  test_data = await yaml_load(dir=case_dir, file=case_name)  result = BXMDict({  'case_dir': os.path.dirname(case_name),  'api': test_data.args[1].replace('/', '_'),  })  if isinstance(test_data.kwargs, list):  for index, each_data in enumerate(test_data.kwargs):  step_name = each_data.pop('caseName')  r = await HTTP(session, domain, *test_data.args, **each_data)  r.update({'case_name': step_name})  result.setdefault('responses', BXMList()).append({  'response': r,  'validator': test_data.validator[index]  })  else:  step_name = test_data.kwargs.pop('caseName')  r = await HTTP(session, domain, *test_data.args, **test_data.kwargs)  r.update({'case_name': step_name})  result.setdefault('responses', BXMList()).append({  'response': r,  'validator': test_data.validator  })  return result

事件循环负责执行协程并返回结果，在最后的结果收集中，我用测试用例目录来对结果进行了分类，这为接下来的自动生成 pytest 认可的测试用例打下了良好的基础。

        def main(test_cases):  """  事件循环主函数，负责所有接口请求的执行  :param test_cases:  :return:  """  loop = asyncio.get_event_loop()  semaphore = asyncio.Semaphore(bxmat.semaphore)  # 需要处理的任务  # tasks = [asyncio.ensure_future(one(case_name=test_case, semaphore=semaphore)) for test_case in test_cases]  task = loop.create_task(entrace(test_cases, loop, semaphore))  # 将协程注册到事件循环，并启动事件循环  try:  # loop.run_until_complete(asyncio.gather(*tasks))  loop.run_until_complete(task)  finally:  loop.close()  return task.result()

第二部分

动态生成 pytest 认可的测试用例

首先说明下 pytest 的运行机制，pytest 首先会在当前目录下找 conftest.py
文件，如果找到了，则先运行它，然后根据命令行参数去指定的目录下找 test 开头或结尾的 .py 文件，如果找到了，如果找到了，再分析
fixture，如果有 session 或 module 类型的，并且参数 autotest=True 或标记了
pytest.mark.usefixtures(a…)，则先运行它们；再去依次找类、方法等，规则类似。大概就是这样一个过程。

可以看出，pytest 测试运行起来的关键是，必须有至少一个被 pytest
发现机制认可的testxx.py文件，文件中有TestxxClass类，类中至少有一个def testxx(self)方法。

现在并没有任何 pytest 认可的测试文件，所以我的想法是先创建一个引导型的测试文件，它负责让 pytest
动起来。可以用pytest.skip()让其中的测试方法跳过。然后我们的目标是在 pytest
动起来之后，怎么动态生成用例，然后发现这些用例，执行这些用例，生成测试报告，一气呵成。

        # test_bootstrap.py  import pytest  class TestStarter(object):  def test_start(self):  pytest.skip(' 此为测试启动方法 , 不执行 ')

我想到的是通过 fixture，因为 fixture 有 setup 的能力，这样我通过定义一个 scope 为 session 的 fixture，然后在
TestStarter 上面标记 use，就可以在导入 TestStarter 之前预先处理一些事情，那么我把生成用例的操作放在这个 fixture
里就能完成目标了。

        # test_bootstrap.py  import pytest  @pytest.mark.usefixtures('te', 'test_cases')  class TestStarter(object):  def test_start(self):  pytest.skip(' 此为测试启动方法 , 不执行 ')

pytest 有个--rootdir参数，该 fixture
的核心目的就是，通过--rootdir获取到目标目录，找出里面的.yml测试文件，运行后获得测试数据，然后为每个目录创建一份testxx.py的测试文件，文件内容就是content变量的内容，然后把这些参数再传给pytest.main()方法执行测试用例的测试，也就是在
pytest 内部再运行了一个 pytest！最后把生成的测试文件删除。注意该 fixture 要定义在conftest.py里面，因为 pytest
对于conftest中定义的内容有自发现能力，不需要额外导入。

        # conftest.py  @pytest.fixture(scope='session')  def test_cases(request):  """  测试用例生成处理  :param request:  :return:  """  var = request.config.getoption("--rootdir")  test_file = request.config.getoption("--tf")  env = request.config.getoption("--te")  cases = []  if test_file:  cases = [test_file]  else:  if os.path.isdir(var):  for root, dirs, files in os.walk(var):  if re.match(r'\w+', root):  if files:  cases.extend([os.path.join(root, file) for file in files if file.endswith('yml')])  data = main(cases)  content = """  import allure  from conftest import CaseMetaClass  @allure.feature('{}接口测试 ({}项目)')  class Test{}API(object, metaclass=CaseMetaClass):  test_cases_data = {}  """  test_cases_files = []  if os.path.isdir(var):  for root, dirs, files in os.walk(var):  if not ('.' in root or '__' in root):  if files:  case_name = os.path.basename(root)  project_name = os.path.basename(os.path.dirname(root))  test_case_file = os.path.join(root, 'test_{}.py'.format(case_name))  with open(test_case_file, 'w', encoding='utf-8') as fw:  fw.write(content.format(case_name, project_name, case_name.title(), data.get(root)))  test_cases_files.append(test_case_file)  if test_file:  temp = os.path.dirname(test_file)  py_file = os.path.join(temp, 'test_{}.py'.format(os.path.basename(temp)))  else:  py_file = var  pytest.main([  '-v',  py_file,  '--alluredir',  'report',  '--te',  env,  '--capture',  'no',  '--disable-warnings',  ])  for file in test_cases_files:  os.remove(file)  return test_cases_files

可以看到，测试文件中有一个TestxxAPI的类，它只有一个test_cases_data属性，并没有testxx方法，所以还不是被
pytest 认可的测试用例，根本运行不起来。那么它是怎么解决这个问题的呢？答案就是CaseMetaClass。

        function_express = """  def {}(self, response, validata):  with allure.step(response.pop('case_name')):  validator(response,validata)"""  class CaseMetaClass(type):  """  根据接口调用的结果自动生成测试用例  """  def __new__(cls, name, bases, attrs):  test_cases_data = attrs.pop('test_cases_data')  for each in test_cases_data:  api = each.pop('api')  function_name = 'test' + api  test_data = [tuple(x.values()) for x in each.get('responses')]  function = gen_function(function_express.format(function_name),  namespace={'validator': validator, 'allure': allure})  # 集成 allure  story_function = allure.story('{}'.format(api.replace('_', '/')))(function)  attrs[function_name] = pytest.mark.parametrize('response,validata', test_data)(story_function)  return super().__new__(cls, name, bases, attrs)

CaseMetaClass是一个元类，它读取 test_cases_data 属性的内容，然后动态生成方法对象，每一个接口都是单独一个方法，在相继被
allure 的细粒度测试报告功能和 pytest
提供的参数化测试功能装饰后，把该方法对象赋值给test+api的类属性，也就是说，TestxxAPI在生成之后便有了若干testxx的方法，此时内部再运行起
pytest，pytest 也就能发现这些用例并执行了。

        def gen_function(function_express, namespace={}):  """  动态生成函数对象 , 函数作用域默认设置为 builtins.__dict__，并合并 namespace 的变量  :param function_express: 函数表达式，示例 'def foobar(): return "foobar"'  :return:  """  builtins.__dict__.update(namespace)  module_code = compile(function_express, '', 'exec')  function_code = [c for c in module_code.co_consts if isinstance(c, types.CodeType)][0]  return types.FunctionType(function_code, builtins.__dict__)

在生成方法对象时要注意 namespace 的问题，最好默认传builtins.__dict__，然后自定义的方法通过 namespace 参数传进去。

后续（yml测试文件自动生成）

至此，框架的核心功能已经完成了，经过几个项目的实践，效果完全超过预期，写起用例来不要太爽，运行起来不要太快，测试报告也整的明明白白漂漂亮亮的，但我发现还是有些累，为什么呢？

我目前做接口测试的流程是，如果项目集成了 swagger，通过 swagger
去获取接口信息，根据这些接口信息来手工起项目创建用例。这个过程很重复很繁琐，因为我们的用例模板已经大致固定了，其实用例之间就是一些参数比如目录、用例名称、method
等等的区别，那么这个过程我觉得完全可以自动化。

因为 swagger 有个网页，我可以去提取关键信息来自动创建 .yml 测试文件，就像搭起架子一样，待项目架子生成后，我再去设计用例填传参就可以了。

于是我试着去解析请求 swagger 首页得到的 HTML，然后失望的是并没有实际数据，后来猜想应该是用了
ajax，打开浏览器控制台的时，我发现了api-docs的请求，一看果然是 json 数据，那么问题就简单了，网页分析都不用了。

        import re  import os  import sys  from requests import Session  template ="""  args:  - {method}  -       - {api}  -     kwargs:  -       -  -         caseName: {caseName}  -         {data_or_params}:  -             {data}  -     validator:  -       -  -         json:  -           successed: True  -     """  -       -       -     def auto_gen_cases(swagger_url, project_name):  -         """  -         根据 swagger 返回的 json 数据自动生成 yml 测试用例模板  -         :param swagger_url:  -         :param project_name:  -         :return:  -         """  -         res = Session().request('get', swagger_url).json()  -         data = res.get('paths')  -       -         workspace = os.getcwd()  -       -         project_ = os.path.join(workspace, project_name)  -       -         if not os.path.exists(project_):  -             os.mkdir(project_)  -       -         for k, v in data.items():  -             pa_res = re.split(r'[/]+', k)  -             dir, *file = pa_res[1:]  -       -             if file:  -                 file = ''.join([x.title() for x in file])  -             else:  -                 file = dir  -       -             file += '.yml'  -       -             dirs = os.path.join(project_, dir)  -       -             if not os.path.exists(dirs):  -                 os.mkdir(dirs)  -       -             os.chdir(dirs)  -       -             if len(v) > 1:  -                 v = {'post': v.get('post')}  -             for _k, _v in v.items():  -                 method = _k  -                 api = k  -                 caseName = _v.get('description')  -                 data_or_params = 'params' if method == 'get' else 'data'  -                 parameters = _v.get('parameters')  -       -                 data_s = ''  -                 try:  -                     for each in parameters:  -                         data_s += each.get('name')  -                         data_s += ': \n'  -                         data_s += ' ' * 8  -                 except TypeError:  -                     data_s += '{}'  -       -             file_ = os.path.join(dirs, file)  -       -             with open(file_, 'w', encoding='utf-8') as fw:  -                 fw.write(template.format(  -                     method=method,  -                     api=api,  -                     caseName=caseName,  -                     data_or_params=data_or_params,  -                     data=data_s  -                 ))  -       -             os.chdir(project_)  -

现在要开始一个项目的接口测试覆盖，只要该项目集成了
swagger，就能秒生成项目架子，测试人员只需要专心设计接口测试用例即可，我觉得对于测试团队的推广使用是很有意义的，也更方便了我这样的懒人。

以上，供大家参考，请多指正！

来霍格沃兹测试开发学社，学习更多软件测试与测试开发的进阶技术，知识点涵盖web自动化测试 app自动化测试、接口自动化测试、测试框架、性能测试、安全测试、持续集成/持续交付/DevOps，测试左移、测试右移、精准测试、测试平台开发、测试管理等内容，课程技术涵盖bash、pytest、junit、selenium、appium、postman、requests、httprunner、jmeter、jenkins、docker、k8s、elk、sonarqube、jacoco、jvm-sandbox等相关技术，全面提升测试开发工程师的技术实力
QQ交流群：484590337
公众号 TestingStudio
更多内容欢迎访问 https://ceshiren.com
测试人社区
视频资料领取：https://qrcode.testing-studio.com/f?from=CSDN&url=https://ceshiren.com/t/topic/15844
点击查看更多信息