接口自动化代码设计结构

基本介绍

利用python代码实现接口自动化，通过编写数据配置层、工具层/公共层、请求方法层、用例层、报告层将功能进行分层，每层都有自己独立的功能。这里先从最基本简单的单用户独立脚本开始编写。

• 数据配置层：分别存储用户的基本配置、用例数据、请求的参数、期望的数据文件、sql语句(暂不加)。
• 工具层/公共层：将数据配置层中的拼接好的完整数据放在一个二维列表中
• 请求方法层：封装公共的请求方法
• 用例层：调用公共请求方法将测试数据发送给服务器，并断言
• 报告层：存放测试的结果，用于查阅或输出（暂不加）

实战步骤

首先理清我们的每个文件夹所对应的层级和功能，在每个层级中添加数据或代码进行相互的关联，以下是具体的文件层级和功能。

• common：将common作为公共层，来读取相应的excel、ini、json文件，并将完整的数据拼接成一个完整的二维列表。由于需要执行文件所以需要创建成python package。
• data_config：将data_config作为数据配置层，来存储用户的基本配置文件ini，用例数据文件excel，请求的参数文件json，期望数据文件json。由于只是存储文件不需要执行，这里可创建为普通目录。
• report：将report作为报告层，存储输出的报告和测试的结果。（暂时不动）
• request_method：将request_method作为请求方法层，编写公共的请求方法，因此创建成python package
• test_case：将test_case作为用例层，编写将请求的数据的结果与期望数据进行对比，加入断言。

数据配置层

APIAutoTest.xlsx

在此文件中有测试用例的基本数据，编号、模块名称、接口名称、用例标题、用例的等级、请求的方法、请求的媒体类型、请求的路径、用例数据的标题、期望数据的标题。这里存放用例数据的标题和期望数据的标题，将具体的用例数据和期望数据存放在另一个json文件中，这样做的原因是条理清晰、方便修改、方便读取、方便最后实际结果和期望结果进行对比。

config.ini

在此文件中存放用户的基本信息、以ini基本格式[节点名称]下的键值对存放具体的文件信息、通用的域名地址、excel的工作表名称、数据库基本信息。

case_data

在此文件中存放具体的模块-接口-用例数据标题下的请求参数，通过模块名称、接口名称、用例数据标题来确认参数的归属

expect_data

在此文件中存放具体的模块-接口-期望数据标题下的期望数据，通过模块名称、接口名称、期望数据标题来确认参数的归属

工具层/公共层

read_ini.py

整理思路：首先创建class ReadExcel，利用单例模式读取ini文件的路径和对象，然后创建get_file_path方法通过ini文件的file节点下的文件名称来获取数据配置层当中的文件路径，创建get_host方法通过ini文件的host节点下的键值来获取请求的域名，创建get_table_name方法通过ini文件下的table节点中的键值来获取excel的工作表名称。

创建类和单例

import configparser
import os

class ReadIni:
    instance = None
    def __new__(cls, *args, **kwargs):
        if cls.instance is None:
            # 1 读取ini文件
            # 1.1 获取data_config目录的路径
            cls.data_config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data_config")
            # 1.2 获取ini文件的路径
            ini_path = os.path.join(cls.data_config_path, "config.ini")
            # 1.3 创建Configparser对象
            cls.conf = configparser.ConfigParser()
            # 1.4 Configparser对象调用read方法读取ini文件
            cls.conf.read(ini_path, encoding="utf-8")
            # 获取对象的内存地址
            cls.instance = super().__new__(cls, *args, **kwargs)
        return cls.instance

创建方法get_file_path

    def get_file_path(self, key):
        """根据key，获取file节点下key所对应文件的路径"""
        try:
            # 使用Configparser对象的get方法获取key对应的文件名称
            file_name = self.conf.get("file", key)
        except Exception as e:
            # 记录错误
            print(f"执行ReadIni下的get_file_path方法，传入的{key},获取数据时，报错，错误为：{type(e)},错误的描述为：{e}")
            raise e
        else:
            # 拼接文件的路径，再返回
            return os.path.join(self.data_config_path, file_name)

创建方法get_host

    def get_host(self, key):
        """根据key，获取host节点下被测系统的域名"""
        try:
            # 使用Configparser对象的get方法获取key对应的值
            value = self.conf.get("host", key)
        except Exception as e:
            print(f"执行ReadIni下的get_host方法，传入的{key},获取数据时，报错，错误为：{type(e)},错误的描述为：{e}")
            raise e
        else:
            return value

创建方法get_tabl_name

    def get_table_name(self, key):
        """根据key，获取table节点下工作表的名称"""
        try:
            # 使用Configparser对象的get方法获取key对应的值
            value = self.conf.get("table", key)
        except Exception as e:
            print(f"执行ReadIni下的get_table_name方法，传入的{key},获取数据时，报错，错误为：{type(e)},错误的描述为：{e}")
            raise e
        else:
            return value

read_json.py

整理思路：读取json文件，并将json文件的数据序列化为python的对象再返回。

创建方法read_json

import json
import os

from APIAutoTest_v1.common.read_ini import ReadIni


def read_json(file_path):
    """读取json文件，将json文件的数据序列化为python对象再返回"""
    # 校验传入的json文件路径
    if isinstance(file_path, str) and file_path.lower().endswith(".json") and os.path.isfile(file_path):
        try:
            # 获取json文件的对象
            with open(file_path, mode="r", encoding="utf-8") as f:
                # 使用json.loads/load将json文件的内容序列化为python对象，再返回
                return json.load(f)
        except Exception as e:
            print(f"将json文件的内容序列化为python对象时报错")
            raise e
    else:
        raise FileExistsError("文件路径错误")

read_excel.py

整理思路：首先初始化excel连接对象，通过调用read_ini.py中的get_file_path方法获取excel文件的路径和通过get_file_path方法获取工作表的名称，最后加载工作簿和获取工作表。创建__get_cell_value来获取获取指定列号和行号的单元格数据,通过__get_cell_value来创建获取用例的请求方法、媒体类型、请求的url、模块名称、接口名称，最后创建 case_data方法来获取完整的用例数据。

ReadExcel

import openpyxl

from APIAutoTest_v1.common.read_ini import ReadIni
from APIAutoTest_v1.common.read_json import read_json

class ReadExcel:
    def __init__(self):
        """获取excel的路径，加载excel的工作簿，获取excel的工作表"""
        # 1：创建ReadIni对象
        self.ini = ReadIni()
        # 2：ReadIni对象调用get_file_path获取excel的路径
        excel_path = self.ini.get_file_path("excel")
        # 3: ReadIni对象调用get_table_name获取excel的工作表的名称
        table_name = self.ini.get_table_name("name")
        try:
            # 4: 加载excel的工作簿
            wb = openpyxl.load_workbook(excel_path)
            # 5: 获取工作表
            self.ws = wb[table_name]
        except Exception as e:
            print(f"加载excel的工作簿或获取工作表时报错，请查看配置文件的信息, 错误的类型为：{type(e)}， 错误的描述：{e}")
            raise e

__get_cell_value

    def __get_cell_value(self, column: str, row: int)->str|None:
        """指定获取某个单元格数据， 工作表["列号行号"].value"""
        try:
            # 获取指定列号和行号的单元格数据
            value = self.ws[column + str(row)].value
        except Exception as e:
            print(f"指定获取某个单元格数据，报错，请查看传入的列号和行号，传入的列号为：{column}, 行号为：{row}")
            raise e
        else:
            # 判断单元格数据是否为字符串，如果是字符串，再判断去掉前后空白字符之后字符串的长度要大于0
            # 判断通过之后，返回去掉前后空白字符的字符串
            if isinstance(value, str) and len(value.strip()) > 0:
                return value.strip()
            # 如果单元格的数据不满足上面的条件，不进行任何处理，使用方法的默认返回值None

case_method

    def case_method(self, row):
        """根据行，获取用例的请求方法"""
        # 调用__get_cell_value获取数据
        return self.__get_cell_value("f", row)

case_mime

def case_mime(self, row):
    """根据行，获取用例的媒体类型"""
    # 调用__get_cell_value获取数据
    value = self.__get_cell_value("g", row)
    # 判断媒体类型的数据是否不为None时，将媒体类型的值转小写再返回
    if value:
        return value.lower()

case_url

    def case_url(self, row):
        """根据行，获取用例请求的url"""
        # 调用__get_cell_value获取数据
        path = self.__get_cell_value("h", row)
        # 判断用例请求的路径是否不为None，不为None时，将请求的路径和域名进行拼接完整的url
        if path:
            # 使用ReadIni的对象调用get_host获取被测系统的域名
            host = self.ini.get_host("host")
            return host + path

module_name

    def module_name(self, row):
        """根据行，获取模块名称"""
        # 调用__get_cell_value获取数据
        return self.__get_cell_value("b", row)

api_name

    def api_name(self, row):
        """根据行，获取接口名称"""
        # 调用__get_cell_value获取数据
        return self.__get_cell_value("C", row)

case_data

    def case_data(self, row):
        """根据行，获取用例数据"""
        # 1: 根据行，获取用例数据的key
        case_data_key = self.__get_cell_value("i", row)
        # 2: 根据行，获取模块名称
        module = self.module_name(row)
        # 3: 根据行，获取接口名称
        api = self.api_name(row)
        # 4：判断用例数据的key、模块名称、接口名称同时不为None时，才提取用例数据
        if case_data_key and module and api:
            # 5：使用ReadIni的对象调用get_file_path获取用例json文件的路径
            case_data_path = self.ini.get_file_path("case")
            # 6：调用read_json函数读取用例json文件，并获取到用例数据对应的python对象
            case_data_dict = read_json(case_data_path)
            # 7: 根据用例数据的key、模块名称、接口名称提取用例数据，提取成功之后，返回数据
            try:
                # print(data["模块名称"]["接口名称"]["用例数据的key"])
                return case_data_dict[module][api][case_data_key]
            except Exception as e:
                print(f"请检查用例数据的key={case_data_key}、模块名称={module}、接口名称={api},是否一致，对应的行号为：{row}, ")
                raise e

expect_data

    def expect_data(self, row):
        """根据行，获取期望数据"""
        # 1：获取期望数据的key
        expect_data_key = self.__get_cell_value("j", row)
        # 2：获取模块名称
        module = self.module_name(row)
        # 3：获取接口名称
        api = self.api_name(row)
        # 4：判断期望数据的key、模块名称、接口名称同时不为None时，就提取期望数据
        if expect_data_key and module and api:
            # 5：使用ReadIni的对象调用get_file_path获取期望数据json文件的路径
            expect_data_path = self.ini.get_file_path("expect")
            # 6：调用read_json函数读取期望数据json文件，
            expect_data_dict = read_json(expect_data_path)
            # 7：提取期望数据
            try:
                return expect_data_dict[module][api][expect_data_key]
            except Exception as e:
                print(f"请检查用期望数据的key={expect_data_key}、模块名称={module}、接口名称={api},是否一致，对应的行号为：{row}, ")
                raise e

get_data

    def get_data(self):
        """将测试使用的数据存放在一个二维列表中"""
        # 创建一个空列表，用于存放所有的测试数据
        list_data = []
        # 循环取出每行的测试数据，并将每行的测试数据存放在一个列表中，再将这个列表追加到前进创建的空列表中
        for row in range(2, self.ws.max_row + 1):
            method = self.case_method(row)   # 请求方法
            mime = self.case_mime(row)   # 媒体类型
            url = self.case_url(row)  # url
            data = self.case_data(row)  # 用例数据
            expect = self.expect_data(row) # 期望数据
            # 过滤空行
            if method and url and expect:
                list_data.append([method, mime, url, data, expect])
        else:
            return list_data

请求方法层

整理思路：首先创建class RequestMethod类，设定两个常量的媒体类型，初始化Session对象，配置登录后获取token，最后将token更新到session对象的haders中。创建request方法并判断在各种情况下的请求类型。

request_method.py

import requests
from APIAutoTest_v1.common.read_ini import ReadIni

class RequestMethod:
    mime_json = {"Content-type": "Application/json"}
    mime_form = {"Content-type": "Application/x-www-form-urlencoded"}

    def __init__(self):
        """关联状态"""
        # 创建Session对象
        self.bpm_session = requests.sessions.Session()
        # 配置登录数据
        login_url = ReadIni().get_host("host") + "/auth"
        login_data = {"username": "admin", "password": "VpCbOu6UsClgnlf/KXkmaR4fD95X75kwitphCJu01b7HfX5rs6gox6/j5722PHBRrPFTeJf8tZV91htWoVb65CLIjUM2X6jgQPQbGJoyY/z+PZwO4oSkklG3ywTG00cdgP+X1eebJjfPdIFZ8nEDnOTyDYv/tVq4Z+Zjd9KxQyI="}
        # 发送请求获取token，并将token更新到Session对象的headers中
        self.bpm_session.headers["Authorization"] = f"Bearer {self.bpm_session.post(url=login_url, json=login_data).json().get('token')}"

    def request(self, req_method, req_url, req_mime, case_data):
        """
        判断媒体类型和请求的用例数据类型使用不同的关键字传参
        :param req_method: 请求方法
        :param req_url: 请求的url
        :param req_mime: 请求的媒体类型
        :param case_data: 请求的用例数据
        :return: Response type
        """
        # 判断媒体类型是否为application/json或者为json
        if req_mime == "application/json" or req_mime == "json":
            # 请求的数据类型是否为字符串，如果是，使用data传参，更改请求的媒体类型为application/json，如果不是字符串使用json传参
            if isinstance(case_data, str):
                # 发送请求
                return self.bpm_session.request(method=req_method, url=req_url, data=case_data, headers=self.mime_json)
            else:
                # 发送请求
                return self.bpm_session.request(method=req_method, url=req_url, json=case_data, headers=self.mime_json)

        # 判断媒体类型是否为Application/x-www-form-urlencoded或者为form
        elif req_mime == "application/x-www-form-urlencoded" or req_mime == "form":
            # 使用data传参，更改请求的媒体类型为Application/x-www-form-urlencoded
            return self.bpm_session.request(method=req_method, url=req_url, data=case_data, headers=self.mime_form)

        # 判断媒体类型是否为None，如果为None，可能在地址栏中传参，也有可能没有传参
        elif req_mime is None:
            # 判断请求的数据是否不为None，如果不为None，使用params传参，如果为None表示没有传参。
            if case_data:
                return self.bpm_session.request(method=req_method, url=req_url, params=case_data)
            else:
                return self.bpm_session.request(method=req_method, url=req_url)

        else:
            raise ValueError("媒体类型错误")

用例层-α

整理思路：第一种不使用pytest进行请求，在用例层中调用请求方法层中的方法和数据，然后遍历二维列表来进行接口测试，最后通过期望数据和实际结果进行断言。

bak_test_bpm.py

from APIAutoTest_v1.common.read_excel import ReadExcel
from APIAutoTest_v1.request_method.request_method import RequestMethod

class TestBPM:
    def test_bpm(self):
        # 获取测试数据
        excel = ReadExcel()
        datas = excel.get_data()

        # 创建RequestMethod类对象
        req = RequestMethod()

        for data in datas:
            res = req.request(req_method=data[0], req_url=data[2], req_mime=data[1], case_data=data[3])
            try:
                for key in data[-1]:
                    print("期望数据>>>", data[-1])
                    print("服务器返回数据>>>", res.json())
                    assert data[-1][key] == res.json().get(key)
            except AssertionError:
                print("断言失败")
                print("+" * 100)
            else:
                print("断言成功")
                print("+" * 100)

执行结果

用例层-β

整理思路：通过加入pytest参数化来进行请求。当运行当运行 test_bpm.py 时，Pytest 会 首先 自动发现并加载同级或父级目录下的 conftest.py 文件创建resqust对象在返回值给test_bpm.py进行请求和断言.

test_bpm.py

import pytest

from APIAutoTest_v1.common.read_excel import ReadExcel

# 获取数据
datas = ReadExcel().get_data()

class TestBPM:

    @pytest.mark.parametrize("method, mime, url, data, expect", datas)
    def test_bpm(self, req_fix, method, mime, url, data, expect):
        # print(method, mime, url, data, expect)
        res = req_fix.request(req_method=method, req_url=url, req_mime=mime, case_data=data)
        print("-"*100)
        print(res.text)
        # assert
        for key in expect:
            assert expect[key] == res.json().get(key)

conftest.py

import pytest

from APIAutoTest_v1.request_method.request_method import RequestMethod


@pytest.fixture(scope="session")
def req_fix():
    """创建RequestMethod类对象"""
    req = RequestMethod()
    yield req
    req.bpm_session.close()

执行结果