多测师是一家拥有先进的教学理念,强大的师资团队,业内好评甚多的接口自动化测试培训机构!

17727591462

联系电话

您现在所在位置:接口自动化测试培训 > 新闻资讯

接口自动化测试框架开发实例

更新时间:2021-10-21 02:26:24 作者:admin 浏览:299

接口自动化测试框架开发实例

  近期准备优先做接口测试的覆盖,为此需要开发一个测试框架,经过思考,这次依然想做点儿不一样的东西。

  接口测试是比较讲究效率的,测试人员会希望很快能得到结果反馈,然而接口的数量一般都很多,而且会越来越多,所以提高执行效率很有必要。

  接口测试的用例其实也可以用来兼做简单的压力测试,而压力测试需要并发。

  接口测试的用例有很多重复的东西,测试人员应该只需要关注接口测试的设计,这些重复劳动最好自动化来做。

  接口测试的用例应该尽量简洁,最好用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)

  可以看到,测试用例还支持一定的模板语法,如

  (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 += ': '

  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_)

  以上是关于接口自动化测试框架开发实例的分享,由多测师亲自撰写。https://www.aichudan.com/

联系电话

17727591462

返回顶部