欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

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

程序员文章站 2022-03-12 19:15:38
前言趁着这个周末闲来无事,简单的开发了一个接口自动化测试框架。由于我本人也是接口自动化测试的新手,如有不正确的地方请多多指教。流程说明图这张图是我的一些设计思路。在yaml文件中管理相关的数据即可实现接口测试。框架体系介绍目录/文件说明是否为python包apiData存放测试信息和用例的yaml文件目录basic基类包,封装requests,json等常用方法是common公共类,封装读取yaml文件,cookies等常用方法是confi...

前言

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

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

流程说明图

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

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

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

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

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

框架体系介绍

目录/文件 说明 是否为python
apiData 存放测试信息和用例的yaml文件目录
basic 基类包,封装requestsjson等常用方法
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__ = '1084502012@qq.com'

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__ = '1084502012@qq.com'

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__ = '1084502012@qq.com'

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

运行结果

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

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

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

本文地址:https://blog.csdn.net/oHuaXin1234/article/details/107300489