pytest+requests+excel+allure接口测试，多接口依赖处理完整demo

写在前面

本demo是基于基于 极客时间 查询账户下消费订单记录的case，该接口的请求需要登录态的cookie及订单列表中已知的订单号作为依赖参数去请求，还算比较典型。需要先了解 jsonpath语法和 契约测试语法，如要尝试运行，请自行注册极客时间账号。

目录结构

见下图：

设计思想

见图：

具体代码

configs

存放配置文件的目录，整个项目中的各种配置文件都仍在这里面统一管理，维护起来也比较集中好找

baseconf.yaml

存放基础的配置项，比如数据库、服务器账号等，在需要用的地方直接读取它就行

test_db:
  host: xxx
  user: xxx
  pwd: xxx
test_server:
  host: xxx
  port: xxx
  user: xxx
  pwd: xxx
  tout: 20

presets.yaml

测试用例基础信息配置文件，相当于是 test_case 文件中具体 .xlsx文件的索引，可以设定对应的 .xlsx中测试用例执行的级别，并且存放接口请求需要的默认 headers 等信息

test_1:
  logLevel: info
  filePath: 'test_case'  # 模版见 ../test_case/test_1.xlsx
  file_name: 'test_1.xlsx'
  caesLevel: [2]  # 执行_case.xlsx文件Level列值与之匹配的case，空列表（[]）则执行全部
  headers: '{
            "Content-Type": "application/json",
            "Origin": "https://b.geekbang.org",
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.63 Safari/537.36"
            }'
  payload: '{}'
test_2:
  logLevel: info
  filePath: 'test_case'  # 模版见 ../test_case/test_3.xlsx
  file_name: 'test_2.xlsx'
  caesLevel: [2]  # 执行_case.xlsx文件Level列值与之匹配的case，空列表（[]）则执行全部
  headers: '{
    "Content-Type": "application/json",
    "Origin": "https://account.geekbang.com",
    "Referer": "https://account.geekbang.com/login?redirect=https://time.geekbang.com/",
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.63 Safari/537.36"
    }'
  payload: '{}'

test_case

测试用例管理文件，构思是不同模块或者接口用一个单独的 .xlsx文件管理，具体粒度可以由个人项目情况而定
test_1.xlsx 和 test_2.xlsx内容格式完全一致，只是用于存放不同的测试用例而已

test_2.xlsx

test_run

测试用例的执行入口

conftest.py

代码运行的主要配置文件，整个自动化运行基本只需要维护该文件就行了，全局变量 OBJ 用于指定要执行的case文件，对应 configs> presets.yaml 中的字典key，例如指定 OBJ = ‘test_2’，则运行 configs> presets.yaml> test_2下面配置中对应的case

# !/usr/local/bin/python3
# coding=utf-8
# @Software: PyCharm
# @File: conftest.py
# @Author: damon 
# @Time: 2023/2/27


import pytest
from utils.util import log
from utils.request_run import RunCase
from utils.get_excel_test_case import group_test_data
from utils.util import get_data_from_yaml, SetUpRequests

OBJ = 'test_2'
DEV = 2  # 1、测试环境 2、生产环境
CONFIG = get_data_from_yaml('configs/presets.yaml')
CONFIG_ = CONFIG[OBJ]  # 获取配置文件中对应的配置信息
OPTIONVALUE = 1
TITLE = {
    "title": "极客时间个人中心",
    "subtitle": "我的订单查询"
}


def pytest_addoption(parser):
    parser.addoption("--dev", action="store", default="default name")
    parser.addoption("--name", action="store", default="default name")


def pytest_generate_tests(metafunc):
    global OPTIONVALUE
    try:
        OPTIONVALUE = metafunc.config.option.dev
    except:
        OPTIONVALUE = 1
    if 'dev' in metafunc.fixturenames and OPTIONVALUE is not None:
        metafunc.parametrize("dev", [OPTIONVALUE])
    option_values = metafunc.config.option.name
    if 'name' in metafunc.fixturenames and option_values is not None:
        metafunc.parametrize("name", [option_values])


@pytest.fixture(scope='class')
def set_up_login():
    print('\n<<<---- 开始执行 ---->>>')
    headers = CONFIG_['headers']
    cases = RunCase(CONFIG_)
    try:
        setup = group_test_data(OBJ, sheet='setup', dev=int(OPTIONVALUE))
    except:
        setup = group_test_data(OBJ, sheet='setup', dev=DEV)
    method, url, payload, header = setup[0][2], setup[0][3], setup[0][4], setup[0][5]
    get_cookie = SetUpRequests()
    header = get_cookie.headers_join(headers, header)
    cookies = get_cookie.get_cookie(payload=payload, headers=header, method=method, url=url)
    yield headers, cases, setup, cookies, log
    print('\n<<<---- 执行完毕 ---->>>')

run_test.py

# !/usr/local/bin/python3
# coding=utf-8
# @Software: PyCharm
# @File: run_test.py
# @Author: damon 
# @Time: 2023/2/27


import os
import allure
import pytest
import urllib3
from utils.util import SetUpRequests
from utils.get_excel_test_case import group_test_data
from test_run.conftest import OBJ,TITLE

case_data = group_test_data(OBJ)

@allure.epic(TITLE["title"])
@allure.feature(TITLE["subtitle"])
class TestRun:
    @pytest.mark.parametrize('title,subtitle,method,url,payload,header,expect,id,rely', case_data)
    def test_run(self, method, url, payload, header, expect, id, title, subtitle, rely, set_up_login):
        headers, cases, setup, cookies,log = set_up_login
        urllib3.disable_warnings()  # 强制取消接口请求过多的告警
        allure.dynamic.story(title)
        allure.dynamic.title(subtitle)
        headersjoin = SetUpRequests()
        header = headersjoin.headers_join(headers, header)
        # log(level="info", message="ID:{0} | Title:{1} | Url：{2}".format(id,title,url),console = True)
        cases.run_case(payload=payload, headers=header, method=method, url=url, expect=expect, cookie=cookies, rely=rely)


if __name__ == '__main__':
    base_dir = '/Users/damon/PycharmProjects/tester_home'
    path = os.path.join(base_dir, 'test_run/run_test.py')
    pytest.main([path])

utils

基础方法归类

get_excel_test_case.py

用 openpyxl 库操作 excel 中具体的测试用例

# !/usr/local/bin/python3
# coding=utf-8
# @Software: PyCharm
# @File: get_excel_test_case.py
# @Author: damon 
# @Time: 2023/2/27


import os
from openpyxl import load_workbook
import json
from utils.util import get_data_from_yaml,Log


log = Log()

BASE_DIR = os.path.abspath(os.path.dirname(os.getcwd()))

class SetUpExcel(object):
    def __init__(self):
        pass

    @staticmethod
    def str_to_list(data):
        # data = '[{"id":5,"key":"id_5"};{"id":6,"key":"id_6"}]'
        data = data[1:-1].split(";")
        l = []
        for i in range(len(data)):
            # log.log("error",data[i])
            l.append(json.loads(data[i]))
        return l

    @staticmethod
    def update_dict(data, title, column, row_id, sheet):
        """
        :param data: dict 类型
        :param title: excel的表头生成字典 { "ID": A,"title": B }
        :param column: excel 列数，第column 列
        :param row_id: excel 行数，第row_id 行
        :param sheet:
        :return:
        """
        column_id = title["{column}".format(column=column)]   '1'
        own_title = sheet[column_id].value  # 获取自定义的列名
        values = title["{column}".format(column=column)]   '{row_id}'.format(row_id=row_id)
        key = str.lower(own_title)
        data.update({key: sheet[values].value})
        return data

    @staticmethod
    def excel_title():
        """
        excel的表头生成字典
        { "ID": A,"title": B }
        """
        values = list(map(chr, range(ord('A'), ord('Z')   1)))
        excel_title = {}
        for keys in range(1, len(values)   1):
            excel_title[str(keys)] = values[keys - 1]
        return excel_title

    @staticmethod
    def open_excel(path):
        """根据配置文件中 LEVEL_ 的值读取数据"""
        try:
            assert os.path.exists(path) == True
            workbook = load_workbook(path)
            return workbook
        except Exception as e:
            raise e

    def relevant(self, data, sheet, title) ->dict:
        """
        :param data:
        :param sheet:
        :param title:
        :return:
        """
        global rely_row_id
        r_data = self.str_to_list(data['rely'])
        rely_list = []
        for x in range(len(r_data)):
            relys = {}
            if not isinstance(r_data[x], dict):
                try:
                    r_data[x] = json.loads(r_data[x])
                except:
                    raise Exception('非法数据类型')
            # 获取关联case所属行 rely_row_id
            for i in range(2, sheet.max_row   1):
                v = "A"   str(i)
                is_id = sheet[v].value
                if is_id == r_data[x]['id']:  # 判断这一行的 A 列为空，则停止遍历
                    rely_row_id = i
                    break
            for y in range(1, sheet.max_column   1):
                self.update_dict(relys, title, y, rely_row_id, sheet)
            relys.update({"key": r_data[x]["key"]})
            rely_list.append(relys)
        data["rely"] = rely_list
        # log.debug(data)
        return data

    def if_rely(self, rely, data_dict, sheet, title):
        if rely:
            data_dict = self.relevant(data_dict, sheet, title)
        return data_dict


class ExcelToCase(SetUpExcel):
    """读取 excel 测试用例并格式化成符合@pytest.mark.parametrize(）装饰器规格的  list"""

    def __init__(self, paths,file_name, sheet, level=None, rows=None):
        super(SetUpExcel, self).__init__()
        self.path = os.path.join(BASE_DIR, paths,file_name)
        self.sheets = sheet
        self.rows = rows
        self.LEVEL_ = level

    def case_data(self, data, sheet, title, i):
        """
        excel 行数据加上 title，转换成dict格式 { title:value }
        :param data:
        :param sheet:
        :param title:
        :param i:
        :return:
        """
        max_column = sheet.max_column   1
        for x in range(1, max_column):
            self.update_dict(data, title, x, i, sheet)
        return data

    def read_excel_xlsx(self):
        """
        根据配置文件中 LEVEL_ 的值读取数据
        """
        title = self.excel_title()
        workbook = self.open_excel(self.path)
        sheet = workbook[self.sheets]
        data_list = []
        title_dict = {}
        for i, v in enumerate(sheet[1]):
            v = str.lower(v.value)
            title_dict[v] = i
        """
        当传递rows 参数时，约定是收集 excel中setup sheet里的数据，
        否则收集 cases sheet里的数据
        """
        if self.rows:
            data_dict = {}
            if self.rows > 2:  # 默认强制测试环境 1:测试，2:生产
                self.rows = 1
            """
            把excel数据处理成字典格式，表头是key,数据是value
            """
            data_dict = self.case_data(data_dict, sheet, title, self.rows   1)
            data_list.append(data_dict)
            workbook.close()
            return data_list
        else:
            for i in range(2, sheet.max_row   1):
                if "id" in title_dict:
                    k = title_dict["id"]
                    c = sheet[1][k].coordinate
                    v = c[0]   str(i)
                    is_none = sheet[v].value
                    if not is_none:  # 判断这一行的 A 列为空，则停止遍历
                        break
                    data_dict = {}
                    data_dict = self.case_data(data_dict, sheet, title, i)
                    """
                    收集excel中 level 字段值与 self.LEVEL_ 匹配的数据
                    如果 self.LEVEL_ == Fales,收集excel中所有数据，否则收集对应 level的数据
                    """
                    if not self.LEVEL_:
                        data_dict = self.if_rely(data_dict['rely'], data_dict, sheet, title)
                        data_list.append(data_dict)
                    else:
                        # 获取 Level 列的值
                        if "level" in title_dict:
                            k1 = title_dict["level"]
                            c1 = sheet[1][k1].coordinate
                            row_id = c1[0]   str(i)  # >> D1\D2\D3
                            level = sheet[row_id].value
                            if level in self.LEVEL_:
                                data_dict = self.if_rely(data_dict['rely'], data_dict, sheet, title)
                                data_list.append(data_dict)
            workbook.close()
            # log.debug(f"{data_dict}")
            return data_list


def group_test_data(project, sheet='cases', dev=None) -> list:
    """
    :param project: 配置文件中的项目
    :param sheet: 默认是测试数据 sheet，前置需要重新赋值
    :param dev: 1、测试环境 2、生产环境
    :return:
    """
    CONF = get_data_from_yaml('configs/presets.yaml')
    CONFIG_ = CONF[project]
    level = CONFIG_['caesLevel']
    case = ExcelToCase(paths=CONFIG_['filePath'],file_name= CONFIG_['file_name'], sheet=sheet, level=level, rows=dev)
    data = case.read_excel_xlsx()
    dataslist = []
    for j in range(len(data)):
        datas = (
            data[j]['title'], data[j]['subtitle'], data[j]['method'],
            data[j]['url'], data[j]['payload'], data[j]['header'],
            data[j]['expect'], data[j]['id'], data[j]['rely']
        )
        dataslist.append(datas)
    return dataslist

request_run.py

用 request 库请求被测试的接口

# !/usr/local/bin/python3
# coding=utf-8
# @Software: PyCharm
# @File: request_run.py
# @Author: damon 
# @Time: 2023/2/27


import json
import allure
import requests
from utils.util import log, CheckJson, SetUpRequests
from jsonpath import jsonpath


class RunCase(object):
    def __init__(self, config):
        self.config = config

    @staticmethod
    def my_rely(data, cookie, payload, headers):
        """
        :param data: [{},{}] 是一个完整的用例格式，并携带依赖的key关键字
        :param cookie: 依赖的cookie
        :param payload:
        :param headers:
        :return:
        """
        global r_value
        if payload and not isinstance(payload, dict):
            payload = json.loads(payload)
        for x in range(len(data)):
            rely_data = data[x]
            if not data[x]:
                log("error", "解析 rely 数据失败了")
            headersjoin = SetUpRequests()
            headers = headersjoin.headers_join(headers, rely_data['header'])
            if headers and not isinstance(headers, dict):
                headers = json.loads(headers)
            rely_payload = rely_data['payload']
            if rely_payload and not isinstance(rely_payload, str):
                rely_payload = json.loads(rely_payload)
            try:
                req = requests.request(method=rely_data['method'], url=rely_data['url'], data=rely_payload,
                                       cookies=cookie,
                                       headers=headers, verify=False)
            except:
                rely_payload = rely_payload.encode('utf-8')
                req = requests.request(method=rely_data['method'], url=rely_data['url'], data=rely_payload,
                                       cookies=cookie,
                                       headers=headers, verify=False)
            req = req.json()
            rely_key_list = rely_data['key']  # 依赖的字段名称
            for key, value in rely_key_list.items():
                try:
                    r_value = jsonpath(req, value)[0]
                except:
                    log("error", "依赖解析异常")

                if key in payload.values():
                    """
                    重新赋值请求参数，将excel 中 Payload 列中依赖Rely列标记字段值赋值成接口返回的实际值
                    """
                    for k, v in payload.items():
                        if v == key:
                            payload[k] = r_value
        return str(payload)

    def run_case(self, payload, headers, method, url, expect, cookie=None, rely=None):
        """
        有接口依赖，先去读全局变量中有没有匹配的参数，有则直接赋值
        """
        if payload and not isinstance(payload, str):
            payload = json.loads(payload)
        if headers and not isinstance(headers, dict):
            headers = json.loads(headers)
        """多接口依赖处理出口"""
        if rely:
            # log.info(f"rely是：\n{rely}\n")
            """
            处理接口依赖，rely 是一个完整的用例格式，并携带依赖的key关键字，
            请求接口后会根据key获取依赖接口的值给业务接口使用
            """
            payload = self.my_rely(rely, cookie, payload, self.config['headers'])
        if payload:
            """payload 变量是str类型，python 默认的str类型是单引号，需要强制转成双引号"""
            payload = payload.replace("'", '"')
        """前置数据处理完毕，执行真正的case"""

        try:
            response = requests.request(method, url, data=payload, headers=headers, cookies=cookie, verify=False)
        except:
            payload = payload.encode('utf-8')
            response = requests.request(method, url, data=payload, headers=headers, cookies=cookie, verify=False)
        allure.attach(url, 'Url')
        actual = response.json()
        """结果断言部分"""
        if response.status_code != 200:
            # 如果失败，日志记录本次请求的信息方便排查问题
            log("info", "[method: {0}\turl: {1}\tdata: {2}\theader: {3}\tcookie{4}]".format(method, url, payload, headers,cookie))
            raise Exception(log("error", "code:{}".format(response.status_code)))
        if expect:
            check = CheckJson()
            check_info = check.check_data(actual, expect)
            assert check_info[0] == True, Exception(log("error", check_info[1]))

util.py

公用方法类的，比如访问数据库、服务器等一些比较通用的方法都放着这里面。目前里面有好几个方法在接口测试项目中是没有被用到的，后续会逐步接进去用起来。

# !/usr/local/bin/python3
# coding=utf-8
# @Software: PyCharm
# @File: util.py
# @Author: damon 
# @Time: 2023/2/27


import json
import logging
import os
import sys
import time
import paramiko
import pymysql
import requests
import yaml
import functools
from pactverify.utils import generate_pact_json_by_response
from pactverify.matchers import PactJsonVerify

paths = os.path.dirname(os.path.abspath(__file__))  # 当前路径
root_path = os.path.dirname(paths)  # 上一层目录

def get_data_from_yaml(files):
    with open(os.path.join(root_path, files), "r", encoding="utf8") as f:
        get_back = yaml.load(f, Loader=yaml.FullLoader)
    return get_back

# 计时器
def log_execution_time(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        res = func(*args, **kwargs)
        end = time.perf_counter()
        print('{} took {} ms'.format(func.__name__, (end - start) * 1000))
        return res
    return wrapper


DBINFO = get_data_from_yaml('configs/baseconf.yaml')


class TestDatebase(object):
    """连接数据库"""

    def __init__(self, host=DBINFO['test_db']['host'], user=DBINFO['test_db']['user'], pwd=DBINFO['test_db']['pwd']):
        self.host = host
        self.user = user
        self.pwd = pwd

    def __connect(self):
        self.__conn = pymysql.connect(host=self.host, user=self.user, password=self.pwd)
        self.__cursor = self.__conn.cursor(cursor=pymysql.cursors.DictCursor)

    def __execute(self, sql):
        self.__connect()
        try:
            # 执行SQL语句
            self.__cursor.execute(sql)
            # print(111)
            # 提交到数据库执行
            self.__conn.commit()
            # 关闭数据库连接
            self.__conn.close()
            self.__cursor.close()
        except:
            # 发生错误时回滚
            self.__conn.rollback()
            # 关闭数据库连接
            self.__conn.close()
            self.__cursor.close()

    def sql(self, sql):
        self.__execute(sql)

    def select(self, sql):
        self.__connect()
        self.__cursor.execute(sql)
        datas = self.__cursor.fetchall()
        # 关闭数据库连接
        self.__cursor.close()
        self.__conn.close()
        # print(group_test_data)
        return datas

def log(level="debug", message="debug log info", console=True, text=True):
    sys.path.append(root_path)
    LOGPATH = os.path.join(root_path, "logs")
    # 如果不存在这个logs文件夹，就自动创建一个
    if not os.path.exists(LOGPATH):
        os.mkdir(LOGPATH)
    # 创建一个FileHandler，用于写到本地
    # fh = TimedRotatingFileHandler(filename=self.logname, when="H", interval=1, backupCount=3)

    # 文件的命名
    global levels
    logname = os.path.join(LOGPATH, "%s.log" % time.strftime("%Y%m%d%H"))
    logger = logging.getLogger()

    level = str.lower(level)
    assert level in ["info", "debug", "warn", "error"], "log 级别有误"

    if level == "info":
        levels = logging.INFO
    elif level == "debug":
        levels = logging.DEBUG
    elif level == "warn":
        levels = logging.WARNING
    elif level == "error":
        levels = logging.ERROR
    logger.setLevel(levels)
    # 日志输出格式
    formatter = logging.Formatter("[%(asctime)s][%(levelname)s]: %(message)s")

    fh, ch = '', ''
    if text:
        fh = logging.FileHandler(logname, "a", encoding='utf-8')  # 追加模式
        fh.setLevel(levels)
        fh.setFormatter(formatter)
        logger.addHandler(fh)
    if console:
        # 创建一个StreamHandler,用于输出到控制台
        ch = logging.StreamHandler()
        ch.setLevel(levels)
        ch.setFormatter(formatter)
        logger.addHandler(ch)
    if level == "info":
        logger.info(message)
    elif level == "debug":
        logger.debug(message)
    elif level == "warn":
        logger.warning(message)
    elif level == "error":
        logger.error(message)
    # 这两行代码是为了避免日志输出重复问题
    if ch:
        logger.removeHandler(ch)
    if fh:
        logger.removeHandler(fh)
        # 关闭打开的文件
        fh.close()

class SetUpRequests(object):
    def __init__(self):
        pass

    @staticmethod
    def get_cookie(method, url, headers, payload):
        """
        :param url:
        :param method:
        :param headers:
        :param payload:
        :return:
        """
        if payload and not isinstance(payload, str):
            payload = json.loads(payload)
        if headers and not isinstance(headers, dict):
            headers = json.loads(headers)
        try:
            response = requests.request(method, url, headers=headers, data=payload)
        except:
            payload = payload.encode('utf-8')
            response = requests.request(method, url, headers=headers, data=payload)
        cookies = response.cookies
        return cookies

    @staticmethod
    def headers_join(base_header, new_header):
        """
        :param base_header: /configs/presets.yanl 配置文件中的默认header
        :param new_header: excel 中维护的header
        :return: 拼接完整的 header
        """
        headers = {}
        ans = isinstance(base_header, dict)
        if not ans:
            headers = json.loads(base_header)
        if not new_header:
            return headers
        if not isinstance(new_header, dict):
            new_header = json.loads(new_header)
        headers.update(new_header)
        return headers

class CheckJson(object):
    @staticmethod
    def check_data(actual, expect, hard_mode=False):
        if isinstance(expect, str):
            expect = eval(expect)
        check_error_list = {}
        # hard_mode默认为true,hard_mode = True时,实际返回key必须严格等于预期key;hard_mode = False时,实际返回key包含预期key即可
        # separator可自定义指定json关键字标识符，默认为$
        json_verify = PactJsonVerify(expect, hard_mode=hard_mode, separator='$')
        # 校验实际返回数据
        json_verify.verify(actual)
        # 校验结果，通过：True，不通过：False
        result = json_verify.verify_result
        if not result:
            error_infos = {
                "key比预期定义少": json_verify.key_less_than_expect_error,
                "key比预期定义多": json_verify.key_more_than_expect_error,
                "值不匹配错误": json_verify.value_not_match_error,
                "类型不匹配错误": json_verify.type_not_match_error,
                "数组长度不匹配错误": json_verify.list_len_not_match_error,
                "枚举不匹配错误": json_verify.enum_not_match_error
            }
            for k, v in error_infos.items():
                if v:
                    check_error_list[k] = v
        return result, check_error_list

    @staticmethod
    def make_pact_json(json):
        """
        :param json: json 格式
        :return: json 契约格式
        """
        pact_json = generate_pact_json_by_response(json, separator='$')
        return pact_json

pytest.ini

[pytest]

addopts = -v -s -q
markers =
    login: this is login case
    user: this is user case

写在最后

走过路过的大佬们多多指教，不胜感激～

这篇好文章是转载于：学新通技术网