pytest+yaml+allure接口自动化测试框架

前言

趁着这个周末闲来无事，简单的开发了一个接口自动化测试框架。

由于我本人也是接口自动化测试的新手，如有不合理或是不正确的地方请多多指教。

流程说明图

这张图是我的一些设计思路。

在yaml文件中管理相关的数据即可实现接口测试。

由于采用的接口是公司的，所以我进行了保密注释。

验证方式是Authorization，没有用到cookies。

框架体系介绍

目录/文件	说明	是否为python包
apiData	存放测试信息和用例的yaml文件目录
basic	基类包，封装requests，json等常用方法	是
common	公共类，封装读取yaml文件，cookies等常用方法	是
config	配置目录，目录配置，allure环境变量配置	是
logs	日志文件
Test	测试用例	是
tools	工具类，日志等	是
pytest.ini	pytest配置文件
run.bat	执行脚本
readme.md	自述文件

配置用例信息

经过excel和yaml的对比，最终我选择了yaml文件管理用例信息。

BusinessInterface.yaml

业务接口测试

添加公告:
  method: post
  route: /property/bulletin_info/
  RequestData:
    data:
      title: 测试公告
      content: 测试公告
      creator_id: 3
      color: '#ffffff'
  expectcode: 201
  regularcheck: '"creator_name":"****"'
  resultcheck: 测试公告

stand_alone_interface.yaml

单个接口测试

登录:
  method: post
  route: /rest-auth/login/
  RequestData:
    data:
      username: ****
      password: ****
  expectcode: 200
  regularcheck: '[A-Za-z0-9-_.]{165}'
  resultcheck: '"username":"******"'

配置测试信息

testInfo.yaml

测试信息配置

test_info: # 测试信息
  url: https:*****
  timeout: 30.0
  cookies:
  headers:
    User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36

读取信息

ApiData.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import sys

sys.path.append('.')
import os
from ruamel import yaml
from config.conf import DATA_DIR


class ApiInfo:
    """接口信息"""

    def __init__(self):
        self.info = os.path.join(DATA_DIR, 'testinfo.yaml')
        self.business_interface = os.path.join(DATA_DIR, 'BusinessInterface.yaml')
        self.stand_alone_interface = os.path.join(DATA_DIR, 'stand_alone_interface.yaml')
        with open(self.info, encoding='utf-8') as f:
            self.info = yaml.safe_load(f)
        with open(self.business_interface, encoding='utf-8') as f:
            self.business = yaml.safe_load(f)
        with open(self.stand_alone_interface, encoding='utf-8') as f:
            self.stand_alone = yaml.safe_load(f)

    def test_info(self, value):
        """测试信息"""
        return self.info['test_info'][value]

    def login_info(self, value):
        """登录信息"""
        return self.stand_alone['登录'][value]

    def case_info(self, name):
        """用例信息"""
        return self.business[name]

    def stand_info(self, name):
        """单个接口"""
        return self.stand_alone[name]


testinfo = ApiInfo()

if __name__ == '__main__':
    print(testinfo.info['登录'])

封装日志

logger.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import sys

sys.path.append('.')
import os
import logging
from config import conf
from datetime import datetime


class Logger:
    def __init__(self, name):
        log_path = self.log_path[:self.log_path.rfind('/')]
        if not os.path.exists(log_path):
            os.makedirs(log_path)
        self.logger = logging.getLogger(name)
        if not self.logger.handlers:
            self.logger.setLevel(logging.DEBUG)

            # 创建一个handler，用于写入日志文件
            fh = logging.FileHandler(self.log_path, encoding='utf-8')
            fh.setLevel(logging.DEBUG)

            # 在控制台输出
            ch = logging.StreamHandler()
            ch.setLevel(logging.INFO)

            # 定义hanler的格式
            formatter = logging.Formatter(self.fmt)
            fh.setFormatter(formatter)
            ch.setFormatter(formatter)

            # 给log添加handles
            self.logger.addHandler(fh)
            self.logger.addHandler(ch)

    @property
    def fmt(self):
        return '%(asctime)s %(levelname)s %(filename)s:%(lineno)d %(message)s'

    @property
    def log_path(self):
        month = datetime.now().strftime("%Y%m")
        return os.path.join(conf.LOG_PATH, '{}.log'.format(month))


log = Logger('root').logger
if __name__ == '__main__':
    log.info("你好")

封装requests

request.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import sys

sys.path.append('.')
__author__ = '[e-mail]'

import json
import allure
import urllib3
import requests
from tools.logger import log
from requests import Response
from requests.status_codes import codes
from requests.exceptions import RequestException
from common.ApiData import testinfo
from common.cookieConf import save_cookie, get_cookie

urllib3.disable_warnings()


class HttpRequest(object):
    """requests方法二次封装"""

    def __init__(self):
        self.timeout = 30.0
        self.r = requests.session()
        self.headers = testinfo.test_info('headers')

    def send_request(self, method: str, route: str, **kwargs):
        """发送请求
        :param method: 发送方法
        :param route: 发送路径
        optional 可选参数
        :param params: 发送参数-"GET"
        :param data: 发送表单-"POST"
        :param json: 发送json-"post"
        :param headers: 头文件
        :param cookies: 验证字典
        :param files: 上传文件,字典：类似文件的对象``
        :param timeout: 等待服务器发送的时间
        :param auth: 基本/摘要/自定义HTTP身份验证
        :param allow_redirects: 允许重定向，默认为True
        :type bool
        :param proxies: 字典映射协议或协议和代理URL的主机名。
        :param stream: 是否立即下载响应内容。默认为“False”。
        :type bool
        :param verify: （可选）一个布尔值，在这种情况下，它控制是否验证服务器的TLS证书或字符串，在这种情况下，它必须是路径到一个CA包使用。默认为“True”。
        :type bool
        :param cert: 如果是字符串，则为ssl客户端证书文件（.pem）的路径
        :return: request响应
        """
        pass
        method = method.upper()
        url = testinfo.test_info('url') + route
        try:
            log.info("Request Url: {}".format(url))
            log.info("Request Method: {}".format(method))
            if kwargs:
                log.info("Request Data: {}".format(kwargs))
            if method == "GET":
                response = self.r.get(url, **kwargs, headers=self.headers, timeout=self.timeout)
            elif method == "POST":
                response = self.r.post(url, **kwargs, headers=self.headers, timeout=self.timeout)
            elif method == "PUT":
                response = self.r.put(url, **kwargs, headers=self.headers, timeout=self.timeout)
            elif method == "DELETE":
                response = self.r.delete(url, **kwargs, headers=self.headers, timeout=self.timeout)
            elif method in ("OPTIONS", "HEAD", "PATCH"):
                response = self.r.request(method, url, **kwargs, headers=self.headers, timeout=self.timeout)
            else:
                raise AttributeError("send request method is ERROR!")
            with allure.step("%s请求接口" % method):
                allure.attach(url, name="请求地址")
                allure.attach(str(response.headers), "请求头")
                if kwargs:
                    allure.attach(json.dumps(kwargs, ensure_ascii=False), name="请求参数")
                allure.attach(str(response.status_code), name="响应状态码")
                allure.attach(str(elapsed_time(response)), name="响应时间")
                allure.attach(response.text, "响应内容")
            log.info(response)
            log.info("Response Data: {}".format(response.text))
            return response
        except RequestException as e:
            log.exception(format(e))
        except Exception as e:
            raise e

    def __call__(self, *args, **kwargs):
        return self.send_request(*args, **kwargs)

    def close_session(self):
        print("关闭会话")
        self.r.close()


def elapsed_time(func: Response, fixed: str = 's'):
    """
    用时函数
    :param func: response实例
    :param fixed: 1或1000 秒或毫秒
    :return:
    """
    try:
        if fixed.lower() == 's':
            second = func.elapsed.total_seconds()
        elif fixed.lower() == 'ms':
            second = func.elapsed.total_seconds() * 1000
        else:
            raise ValueError("{} not in ['s'，'ms']".format(fixed))
        return second
    except RequestException as e:
        log.exception(e)
    except Exception as e:
        raise e


req = HttpRequest()
__all__ = ['req', 'codes']
if __name__ == '__main__':
    r = req('get', '/passport/user/login', verify=False)
    print(r.cookies)
    res = save_cookie(r)
    print(get_cookie(res))

前置条件

conftest.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import sys

sys.path.append('.')
import json
import pytest
from basic.request import req
from common.ApiData import testinfo
from basic.checkresult import check_results


@pytest.fixture(scope='session')
def is_login(request):
    """登录"""
    r = req(testinfo.login_info('method'), testinfo.login_info('route'), **testinfo.login_info('RequestData'))
    result = json.loads(r.text)
    req.headers['Authorization'] = "JWT " + result['token']
    check_results(r, testinfo.stand_info('登录'))

    def fn():
        req.close_session()

    request.addfinalizer(fn)
    return result['token']





if __name__ == '__main__':
    pass

进行测试

无需依赖的接口

test_stand_alone.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys

sys.path.append('.')
__author__ = '[e-mail]'

import pytest
import allure
from basic.request import req
from common.ApiData import testinfo
from basic.checkresult import check_results


@allure.feature("单个API测试")
class TestStandAlone:

    @pytest.mark.parametrize('case', testinfo.stand_alone.values(), ids=testinfo.stand_alone.keys())
    def test_stand_alone_interface(self, case):
        r = req(case['method'], case['route'], **case['RequestData'])
        check_results(r, case)


if __name__ == "__main__":
    pytest.main(['test_business.py'])

无需依赖的接口在测试函数的参数中不传入"is_login"

需要依赖的接口

test_business.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import sys

sys.path.append('.')
import pytest
import allure
from basic.request import req
from common.ApiData import testinfo
from basic.checkresult import check_results


@allure.feature("业务流程API测试")
class TestBusiness:
    @pytest.mark.parametrize('case', testinfo.business.values(), ids=testinfo.business.keys())
    def test_business_interface(self, is_login, case):
        r = req(case['method'], case['route'], **case['RequestData'])
        check_results(r, case)


if __name__ == "__main__":
    pytest.main(['test_business.py'])

需要依赖的接口在测试函数的参数中传入"is_login"参数

校验测试结果

checkresult.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys

sys.path.append('.')
__author__ = '[e-mail]'

import re
import pytest
import allure
from requests import Response


def check_results(r: Response, case_info):
    """检查运行结果"""
    with allure.step("校验返回响应码"):
        allure.attach(name='预期响应码', body=str(case_info['expectcode']))
        allure.attach(name='实际响应码', body=str(r.status_code))
    pytest.assume(case_info['expectcode'] == r.status_code)
    if case_info['resultcheck']:
        with allure.step("校验响应预期值"):
            allure.attach(name='预期值', body=str(case_info['resultcheck']))
            allure.attach(name='实际值', body=r.text)
        pytest.assume(case_info['resultcheck'] in r.text)
    if case_info['regularcheck']:
        with allure.step("正则校验返回结果"):
            allure.attach(name='预期正则', body=case_info['regularcheck'])
            allure.attach(name='响应值', body=str(re.findall(case_info['regularcheck'], r.text)))
        pytest.assume(re.findall(case_info['regularcheck'], r.text))

配置pytest.ini

pytest.ini

[pytest]
addopts = -s -q

配置allure环境变量

APIenv=TEST
APIversion=1.0
APIhost=https://*****
Tester=hoou

执行测试

run.bat

pytest --alluredir allure-results --clean-alluredir

COPY config\environment.properties allure-results

allure generate allure-results -c -o allure-report

allure open allure-report

运行结果

这就是本周末开发的接口自动化测试框架详情。。。里面不全面的地方还是有很多的。

但是简单的接口足够使用了。