pytest+yaml+allure接口自動化測試框架

前言

趁着這個週末閒來無事，簡單的開發了一個接口自動化測試框架。javascript

因爲我本人也是接口自動化測試的新手，若有不合理或是不正確的地方請多多指教。java

流程說明圖

這張圖是個人一些設計思路。python

在yaml文件中管理相關的數據便可實現接口測試。web

採用的接口是智學網網站的API。shell

支持token認證json

框架體系介紹

目錄/文件	說明	是否爲`python`包
apiData	存放測試信息和用例的`yaml`文件目錄
basic	基類包，封裝`requests`，`json`等經常使用方法	是
common	公共類，封裝讀取`yaml`文件，`cookies`等經常使用方法	是
config	配置目錄，目錄配置，allure環境變量配置	是
logs	日誌文件
Test	測試用例	是
tools	工具類，日誌等	是
pytest.ini	pytest配置文件
run.bat	執行腳本
readme.md	自述文件

配置用例信息

通過excel和yaml的對比，最終我選擇了yaml文件管理用例信息。api

BusinessInterface.yaml服務器

業務接口測試cookie

登陸驗證:
  method: post
  route: /loginSuccess/
  RequestData:
    data:
      userId: "{{data}}"
  expectcode: 200
  regularcheck:
  resultcheck: '"result":"success"'

stand_alone_interface.yamlsession

單個接口測試

登陸:
  method: post
  route: /weakPwdLogin/?from=web_login
  RequestData:
    data:
      loginName: 18291900215
      password: dd636482aca022
      code:
      description: encrypt
  expectcode: 200
  regularcheck: '[\d]{16}'
  resultcheck: '"result":"success"'
  extractresult:
    - data

配置測試信息

testInfo.yaml

測試信息配置

test_info: # 測試信息
  url: https://www.zhixue.com
  timeout: 30.0
  headers:
    Accept: application/json, text/javascript, */*; q=0.01
    Accept-Encoding: gzip, deflate, br
    Accept-Language: zh-CN,zh;q=0.9
    Connection: keep-alive
    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
    cookies: aliyungf_tc=AQAAANNdlkvZ2QYAIb2Q221oiyiSOfhg; tlsysSessionId=cf0a8657-4a02-4a40-8530-ca54889da838; isJump=true; deviceId=27763EA6-04F9-4269-A2D5-59BA0FB1F154; 6e12c6a9-fda1-41b3-82ec-cc496762788d=webim-visitor-69CJM3RYGHMP79F7XV2M; loginUserName=18291900215
    X-Requested-With: XMLHttpRequest

讀取信息

ApiData.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import os
from ruamel import yaml
from config.conf import DATA_DIR


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

    def __init__(self):
        self.info_path = os.path.join(DATA_DIR, 'testinfo.yaml')
        self.business_path = os.path.join(DATA_DIR, 'BusinessInterface.yaml')
        self.stand_alone_path = os.path.join(DATA_DIR, 'stand_alone_interface.yaml')

    @classmethod
    def load(cls, path):
        with open(path, encoding='utf-8') as f:
            return yaml.safe_load(f)

    @property
    def info(self):
        return self.load(self.info_path)

    @property
    def business(self):
        return self.load(self.business_path)

    @property
    def stand_alone(self):
        return self.load(self.stand_alone_path)

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

    def login_info(self, value):
        """登陸信息"""
        return self.stand_alone['登陸'].get(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['test_info'])

封裝日誌

logger.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import os
import logging
from config import conf
from datetime import datetime


class Logger:
    def __init__(self):
        self.logger = logging.getLogger()
        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):
        if not os.path.exists(LOG_PATH):
            os.makedirs(LOG_PATH)
        return os.path.join(LOG_PATH, '{}.log'.format(datetime_strftime()))


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

封裝requests

request.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import json
import allure
import urllib3
import requests
from utils.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.RegExp import regexps
from core.serialize import deserialization, serialization
from core.getresult import get_result

urllib3.disable_warnings()

__all__ = ['req', 'codes']


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, extract: str, **kwargs):
        """發送請求
        :param method: 發送方法
        :param route: 發送路徑
        optional 可選參數
        :param extract: 要提取的值
        :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:
                kwargs_str = serialization(kwargs)
                is_sub = regexps.findall(kwargs_str)
                if is_sub:
                    new_kwargs_str = deserialization(regexps.subs(is_sub, kwargs_str))
                    log.info("Request Data: {}".format(new_kwargs_str))
                    kwargs = new_kwargs_str
            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))
            if extract:
                get_result(response, extract)
            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()
if __name__ == '__main__':
    pass

前置條件

conftest.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import json
import pytest
from core.request import req
from common.ApiData import testinfo
from core.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('extractresult'), **testinfo.login_info('RequestData'))
    result = json.loads(r.text)
    check_results(r, testinfo.stand_info('登陸'))
    if 'token' in result:
        req.headers['Authorization'] = "JWT " + result['token']

    def fn():
        req.close_session()

    request.addfinalizer(fn)


if __name__ == '__main__':
    pass

進行測試

無需依賴的接口

test_stand_alone.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pytest
import allure
from core.request import req
from common.ApiData import testinfo
from core.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.get('extractresult'), **case['RequestData'])
        check_results(r, case)
        print(r.cookies)


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

無需依賴的接口在測試函數的參數中不傳入"is_login"

須要依賴的接口

test_business.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import pytest
import allure
from core.request import req
from common.ApiData import testinfo
from core.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.get('extractresult'), **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 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))

接口參數關聯

在接口測試中咱們須要用上一個接口返回的數據，我在思考了兩天以後採起了正則提取的方式來實現此功能，原本想用jinja2模板可是不太會用。

建立正則類

RegExp.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import re
from utils.logger import log
from common.variable import is_vars


class RegExp(object):
    """正則相關類"""

    def __init__(self):
        self.re = re

    def findall(self, string):
        keys = self.re.findall(r"\{{(.*?)}\}", string)
        return keys

    def subs(self, keys, string):
        result = None
        log.info("提取變量：{}".format(keys))
        for i in keys:
            log.info("替換變量：{}".format(i))
            result = self.re.sub(r"\{{%s}}" % i, is_vars.get(i), string)
        log.info("替換結果：{}".format(result))
        return result

    def __call__(self, exp, string):
        return self.re.findall(r'\"%s":"(.*?)"' % exp, string)[0]


regexps = RegExp()

if __name__ == '__main__':
    pass

添加全局變量池

variable.py

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


class Variable(object):
    """全局變量池"""

    def __init__(self):
        super().__init__()

    def set(self, key, value):
        setattr(self, key, value)

    def get(self, key):
        return getattr(self, key)

    def has(self, key):
        return hasattr(self, key)


is_vars = Variable()

if __name__ == '__main__':
    is_vars.set('name', 'hoou')
    print(is_vars.get('name'))

獲取接口的返回值

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pytest
import allure
from utils.logger import log
from requests import Response
from common.variable import is_vars
from common.RegExp import regexps


def get_result(r: Response, extract):
    """獲取值"""
    for i in extract:
        value = regexps(i, r.text)
        log.info("正則提取結果值:{}={}：".format(i, value))
        is_vars.set(i, value)
        pytest.assume(is_vars.has(i))
    with allure.step("提取返回結果中的值"):
        for i in extract:
            allure.attach(name="提取%s" % i, body=is_vars.get(i))

配置pytest.ini

pytest.ini

[pytest]
addopts = -s -q

配置allure環境變量

APIenv=TEST
APIversion=1.0
TestServer=https://www.zhixue.com
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

運行結果

這就是本週末開發的接口自動化測試框架詳情。。。

因爲我才疏學淺，裏面不全面的地方仍是有不少的。好比：

接口以前參數提取和傳遞
密碼MD5加密

等。。。

2020年7月17日晚更新：

已添加接口接口上下文關聯參數提取傳遞

有木有大佬給我點建議或參考實現這些，感激涕零

最後、這個框架對於簡單的跑接口應該足夠使用了。