接口自动化测试框架开发(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*\(.*\))}#39;)
    pattern_function2 = re.compile(r'^\${(.*)}#39;)
    # 匹配取默认值的语法
    pattern_function3 = re.compile(r'^\$\((.*)\)#39;)

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

想看第二篇文章可继续看下文。

（文章来源于霍格沃兹测试学院）