Python+Requests+Pytest+YAML+Allure接口自动化框架

GitHub源码地址（详细注释）：源码

调试项目python自主搭建：附项目源码

一、项目介绍

本项目是基于 Python+Requests+Pytest+YAML+Allure 搭建的 接口自动化测试框架，用于对 REST API 进行测试。
框架的主要特点包括：
模

• 块化设计：采用 分层架构，包括 API 层、业务层、数据层、公共模块、测试用例层，增强可维护性。
• Pytest 测试框架：使用 Pytest 进行测试组织、执行、夹具管理，并提供强大的插件支持。
• Requests 进行 API 测试：封装 HTTP 请求，简化 API 调用流程。
• YAML 管理测试数据：测试数据与代码解耦，支持多环境测试。
• Allure 生成测试报告：提供清晰直观的测试报告，支持测试历史分析。

二、目录结构

PytestDemo/
│── api/                  # 接口封装层，封装用户相关的 API
│   ├── user.py 			# 提供底层 API 访问接口，与后端 API 进行交互
│── common/               # 公共模块（日志、数据库操作、HTTP 请求封装）
│   ├── logger.py         # 日志管理
│   ├── mysql_operate.py  # 数据库操作
│   ├── read_data.py      # 读取 YAML 文件数据
│── config/               # 配置文件（环境变量、测试数据）
│   ├── setting.ini       # MySQL 环境配置
│── core/                 # requests 请求方法封装、关键字返回结果类
│   ├── rest_client.py    # 简化 HTTP 请求的发送，并提供日志记录功能
│   ├── result_base.py    # 结果基类，用于定义接口返回结果的标准结构
│── data/                 # 测试数据文件管理
│   ├── api_test_data.yml 
│   ├── base_data.yml     
│   ├── scenario_test_data.yml     
│── log/                  # 日志文件
│   ├── 20250324.log     
│── operation/            # 业务层（对接口请求进行封装）
│   ├── user.py           # 业务api封装，供测试用例调用。
│── testcases/            # 测试用例
│   ├── report/           # Allure 测试报告
│   ├── api_test/         # 单接口 API 用例集
│   ├── scenario_test/    # 业务链 API 用例集
│── conftest.py           # Pytest 夹具（Fixture）
│── requirements.txt      # 依赖包清单
│── pytest.ini            # Pytest 配置文件

三、框架分层详解

1. api/：接口封装层

import os  # 导入os模块，用于文件和目录操作
from core.rest_client import RestClient  # 从core模块导入RestClient类
from common.read_data import data  # 从common模块导入data类，用于读取数据# 获取当前文件所在的目录的上一级目录路径
BASE_PATH = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
# 构造配置文件setting.ini的完整路径
data_file_path = os.path.join(BASE_PATH, "config", "setting.ini")
# 读取配置文件中的API根URL
api_root_url = data.load_ini(data_file_path)["host"]["api_root_url"]class User(RestClient):  # 继承RestClient类，封装用户相关的接口def __init__(self, api_root_url, **kwargs):"""初始化User类，调用父类的构造方法:param api_root_url: API的根URL:param kwargs: 其他可选参数"""super(User, self).__init__(api_root_url, **kwargs)def list_all_users(self, **kwargs):"""获取所有用户列表:param kwargs: 其他可选参数:return: GET请求返回的响应结果"""return self.get("/users", **kwargs)def list_one_user(self, username, **kwargs):"""获取指定用户名的用户信息:param username: 用户名:param kwargs: 其他可选参数:return: GET请求返回的响应结果"""return self.get("/users/{}".format(username), **kwargs)def register(self, **kwargs):"""用户注册:param kwargs: 包含注册所需参数:return: POST请求返回的响应结果"""return self.post("/register", **kwargs)def login(self, **kwargs):"""用户登录:param kwargs: 包含登录所需参数:return: POST请求返回的响应结果"""return self.post("/login", **kwargs)def update(self, user_id, **kwargs):"""更新用户信息:param user_id: 需要更新的用户ID:param kwargs: 包含更新内容:return: PUT请求返回的响应结果"""return self.put("/update/user/{}".format(user_id), **kwargs)def delete(self, name, **kwargs):"""删除用户:param name: 需要删除的用户名:param kwargs: 其他可选参数:return: POST请求返回的响应结果"""return self.post("/delete/user/{}".format(name), **kwargs)user = User(api_root_url)  # 创建User类的实例，并传入API根URL

作用：
这一层直接与后端 API 交互，实现最基础的接口请求
不处理业务逻辑，只负责请求和返回数据，调用时只需传入参数
继承 RestClient 统一管理 HTTP 请求

2. 公共模块 (common/)

2.1 日志管理 (common/logger.py)

import logging  # 导入logging模块，用于记录日志
import time  # 导入time模块，用于获取当前时间
import os  # 导入os模块，用于文件和目录操作# 获取当前文件所在的目录的上一级目录路径
BASE_PATH = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))# 定义日志文件存放的目录路径
LOG_PATH = os.path.join(BASE_PATH, "log")# 如果日志目录不存在，则创建日志目录
if not os.path.exists(LOG_PATH):os.mkdir(LOG_PATH)class Logger:"""Logger类用于配置和管理日志记录"""def __init__(self):"""初始化Logger类，创建日志记录器"""# 定义日志文件的完整路径，文件名为当前日期.logself.logname = os.path.join(LOG_PATH, "{}.log".format(time.strftime("%Y%m%d")))# 创建一个日志记录器实例，名称为"log"self.logger = logging.getLogger("log")# 设置日志记录器的级别为DEBUG，表示记录所有级别的日志self.logger.setLevel(logging.DEBUG)# 定义日志格式self.formater = logging.Formatter('[%(asctime)s][%(filename)s %(lineno)d][%(levelname)s]: %(message)s')# 创建一个日志文件处理器，用于将日志写入文件self.filelogger = logging.FileHandler(self.logname, mode='a', encoding="UTF-8")# 创建一个控制台处理器，用于在控制台输出日志self.console = logging.StreamHandler()# 设置控制台日志级别为DEBUGself.console.setLevel(logging.DEBUG)# 设置文件日志级别为DEBUGself.filelogger.setLevel(logging.DEBUG)# 为文件日志处理器和控制台日志处理器设置相同的日志格式self.filelogger.setFormatter(self.formater)# 设置控制台日志的格式self.console.setFormatter(self.formater)# 将文件日志处理器添加到日志记录器self.logger.addHandler(self.filelogger)# 将控制台日志处理器添加到日志记录器self.logger.addHandler(self.console)# 创建Logger类的实例，并获取日志记录器
logger = Logger().loggerif __name__ == '__main__':logger.info("---测试开始---")  # 记录INFO级别日志logger.debug("---测试结束---")  # 记录DEBUG级别日志

作用：
封装日志模块，记录测试执行过程。
使用方式：

from common.logger import loggerlogger.info("这是一条测试日志")

2.2 数据库操作（common/mysql_operate.py）

import pymysql  # 导入pymysql库，用于操作MySQL数据库
import os  # 导入os模块，用于路径操作
from common.read_data import data  # 从common模块中导入读取数据的方法
from common.logger import logger  # 从common模块中导入日志记录器# 获取当前文件的上级目录路径
BASE_PATH = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
# 定义配置文件的路径
data_file_path = os.path.join(BASE_PATH, "config", "setting.ini")
# 读取配置文件中的 MySQL 配置信息
data = data.load_ini(data_file_path)["mysql"]# 从配置文件中提取MySQL连接所需的参数，并存储在字典中
DB_CONF = {"host": data["MYSQL_HOST"],        # 数据库主机地址"port": int(data["MYSQL_PORT"]),   # 数据库端口号"user": data["MYSQL_USER"],        # 数据库用户名"password": data["MYSQL_PASSWD"],  # 数据库密码"db": data["MYSQL_DB"]             # 数据库名称
}class MysqlDb():"""MysqlDb 类用于封装 MySQL 数据库的连接和操作方法。"""def __init__(self, db_conf=DB_CONF):"""初始化MySQL连接:param db_conf: 数据库配置字典，包含连接MySQL所需的配置信息"""# 通过字典拆包传递数据库配置信息，建立数据库连接self.conn = pymysql.connect(**db_conf, autocommit=False)# 通过 cursor() 方法创建游标对象，并让查询结果以字典格式输出self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)def __del__(self):"""释放数据库资源，在对象被删除时自动调用"""# 关闭游标self.cur.close()# 关闭数据库连接self.conn.close()def select_db(self, sql):"""执行查询操作:param sql: 要执行的SQL查询语句:return: 查询结果（列表格式）"""# 检查数据库连接是否断开，如果连接断开则重新建立连接self.conn.ping(reconnect=True)# 执行SQL查询self.cur.execute(sql)# 获取查询结果data = self.cur.fetchall()return datadef execute_db(self, sql):"""执行更新、插入或删除操作:param sql: 要执行的SQL语句（更新/插入/删除）"""try:# 检查数据库连接是否断开，如果连接断开则重新建立连接self.conn.ping(reconnect=True)# 执行SQL语句self.cur.execute(sql)# 提交事务self.conn.commit()except Exception as e:# 捕获异常并记录错误日志logger.info("操作MySQL出现错误，错误原因：{}".format(e))# 如果发生异常，回滚事务self.conn.rollback()# 创建MysqlDb对象并使用DB_CONF进行初始化
db = MysqlDb(DB_CONF)

作用：
封装 MySQL 连接，支持数据查询、执行 SQL 语句。
使用方式：

db = MysqlDb()
data = db.select_db("SELECT * FROM user")

2.3 读取配置文件（common/read_data.py）

import yaml  # 用于处理 YAML 文件
import json  # 用于处理 JSON 文件
from configparser import ConfigParser  # 用于处理 INI 配置文件
from common.logger import logger  # 导入日志记录器class MyConfigParser(ConfigParser):"""自定义的 ConfigParser 类，继承自 Python 内置的 ConfigParser。主要用于解决 .ini 文件中的键（option）自动转为小写的问题。"""def __init__(self, defaults=None):# 调用父类 ConfigParser 的构造方法ConfigParser.__init__(self, defaults=defaults)def optionxform(self, optionstr):# 重写 optionxform 方法，使选项名称保持大小写不变return optionstrclass ReadFileData():"""该类用于读取不同格式的文件数据，包括 YAML、JSON 和 INI 配置文件。"""def __init__(self):pass  # 该类的构造方法无特殊初始化操作def load_yaml(self, file_path):"""读取 YAML 文件并解析数据。:param file_path: YAML 文件的路径:return: 解析后的 YAML 数据"""logger.info("加载 {} 文件......".format(file_path))  # 记录日志，指示正在加载 YAML 文件with open(file_path, encoding='utf-8') as f:data = yaml.safe_load(f)  # 使用 safe_load 方法解析 YAML 文件logger.info("读到数据 ==>>  {} ".format(data))  # 记录日志，输出读取的数据return datadef load_json(self, file_path):"""读取 JSON 文件并解析数据。:param file_path: JSON 文件的路径:return: 解析后的 JSON 数据"""logger.info("加载 {} 文件......".format(file_path))  # 记录日志，指示正在加载 JSON 文件with open(file_path, encoding='utf-8') as f:data = json.load(f)  # 解析 JSON 文件logger.info("读到数据 ==>>  {} ".format(data))  # 记录日志，输出读取的数据return datadef load_ini(self, file_path):"""读取 INI 配置文件并解析数据。:param file_path: INI 配置文件的路径:return: 解析后的 INI 数据，转换为字典格式"""logger.info("加载 {} 文件......".format(file_path))  # 记录日志，指示正在加载 INI 文件config = MyConfigParser()  # 创建 MyConfigParser 实例config.read(file_path, encoding="UTF-8")  # 读取 INI 配置文件data = dict(config._sections)  # 将配置文件的 sections 转换为字典格式# print("读到数据 ==>>  {} ".format(data))  # 该行被注释掉，避免调试输出return data# 创建 ReadFileData 类的实例，以便在其他模块中直接使用 data 变量进行文件读取操作
data = ReadFileData()

作用：
读取配置文件（如 setting.ini）、解析 YAML 测试数据，并提供一个统一的接口，使其他模块可以轻松获取配置信息和测试数据。

3. config/：配置管理

[host]
# 测试环境
api_root_url = http://127.0.0.1:5000[mysql]
# MySQL配置
MYSQL_HOST = localhost
MYSQL_PORT = 3306
MYSQL_USER = root
MYSQL_PASSWD = 123456
MYSQL_DB = flask_demo
autocommit: True  # 确保自动提交开启

作用：
存储环境变量、数据库连接信息等
让配置可修改、可复用

4. core/：请求封装层

core/ 目录是请求封装层，主要负责：

• 封装 HTTP 请求方法，提供统一的 API 访问接口（rest_client.py）。
• 定义接口返回的标准格式，方便后续断言（result_base.py）。

4.1 rest_client.py：封装 HTTP 请求

import requests  # 导入requests库，用于发送HTTP请求
import json as complexjson  # 导入json模块并重命名为complexjson，避免与局部变量json冲突
from common.logger import logger  # 导入日志模块，用于记录HTTP请求日志class RestClient():"""RestClient类，封装HTTP请求的方法"""def __init__(self, api_root_url):"""初始化RestClient类:param api_root_url: API的根URL"""self.api_root_url = api_root_url  # 设置API根URLself.session = requests.session()  # 创建一个requests的会话对象，提高请求效率def get(self, url, **kwargs):"""发送GET请求:param url: API接口路径:param kwargs: 其他可选参数（如headers, params等）:return: GET请求的响应对象"""return self.request(url, "GET", **kwargs)def post(self, url, data=None, json=None, **kwargs):"""发送POST请求:param url: API接口路径:param data: 表单数据（可选）:param json: JSON格式数据（可选）:param kwargs: 其他可选参数（如headers等）:return: POST请求的响应对象"""return self.request(url, "POST", data, json, **kwargs)def put(self, url, data=None, **kwargs):"""发送PUT请求:param url: API接口路径:param data: 需要更新的数据:param kwargs: 其他可选参数（如headers等）:return: PUT请求的响应对象"""return self.request(url, "PUT", data, **kwargs)def delete(self, url, **kwargs):"""发送DELETE请求:param url: API接口路径:param kwargs: 其他可选参数（如headers等）:return: DELETE请求的响应对象"""return self.request(url, "DELETE", **kwargs)def patch(self, url, data=None, **kwargs):"""发送PATCH请求:param url: API接口路径:param data: 需要部分更新的数据:param kwargs: 其他可选参数（如headers等）:return: PATCH请求的响应对象"""return self.request(url, "PATCH", data, **kwargs)def request(self, url, method, data=None, json=None, **kwargs):"""统一的请求方法:param url: API接口路径:param method: 请求方法（GET, POST, PUT, DELETE, PATCH）:param data: 表单数据（可选）:param json: JSON格式数据（可选）:param kwargs: 其他可选参数（如headers, params等）:return: 请求的响应对象"""url = self.api_root_url + url  # 拼接完整的URLheaders = dict(**kwargs).get("headers")  # 获取请求头params = dict(**kwargs).get("params")  # 获取请求参数files = dict(**kwargs).get("files")  # 获取文件上传参数cookies = dict(**kwargs).get("cookies")  # 获取cookies参数# 记录请求日志self.request_log(url, method, data, json, params, headers, files, cookies)# 根据请求方法调用相应的requests方法if method == "GET":return self.session.get(url, **kwargs)if method == "POST":return requests.post(url, data=data, json=json, **kwargs)if method == "PUT":if json:data = complexjson.dumps(json)  # 将json对象转换为字符串return self.session.put(url, data=data, **kwargs)if method == "DELETE":return self.session.delete(url, **kwargs)if method == "PATCH":if json:data = complexjson.dumps(json)  # 将json对象转换为字符串return self.session.patch(url, data=data, **kwargs)def request_log(self, url, method, data=None, json=None, params=None, headers=None, files=None, cookies=None,**kwargs):"""记录HTTP请求日志，方便调试和问题排查:param url: 请求的URL:param method: 请求方法（GET, POST, PUT, DELETE, PATCH）:param data: 表单数据（可选）:param json: JSON格式数据（可选）:param params: URL查询参数（可选）:param headers: 请求头（可选）:param files: 上传的文件（可选）:param cookies: 请求的cookies（可选）:param kwargs: 其他可选参数"""logger.info("接口请求地址 ==>> {}".format(url))logger.info("接口请求方式 ==>> {}".format(method))logger.info("接口请求头 ==>> {}".format(complexjson.dumps(headers, indent=4, ensure_ascii=False)))logger.info("接口请求 params 参数 ==>> {}".format(complexjson.dumps(params, indent=4, ensure_ascii=False)))logger.info("接口请求体 data 参数 ==>> {}".format(complexjson.dumps(data, indent=4, ensure_ascii=False)))logger.info("接口请求体 json 参数 ==>> {}".format(complexjson.dumps(json, indent=4, ensure_ascii=False)))logger.info("接口上传附件 files 参数 ==>> {}".format(files))logger.info("接口 cookies 参数 ==>> {}".format(complexjson.dumps(cookies, indent=4, ensure_ascii=False)))

rest_client.py 封装了 GET、POST、PUT、DELETE 请求，提供一个 RestClient 类，让其他模块可以直接调用 HTTP 请求方法，而不需要每次都手写 requests.get() 或 requests.post()。

4.2 result_base.py：封装接口返回结果

class ResultBase:"""结果基类，用于定义接口返回结果的标准结构。该类可以在后续扩展时添加通用的返回数据处理逻辑。"""pass

result_base.py 负责解析接口返回的 JSON 数据，并提供统一的访问方式，让测试用例更加简洁。

5.data/：测试数据管理

作用：
让测试用例数据分离，便于维护
支持数据驱动，自动化测试多个场景

6. operation/：业务封装层

from core.result_base import ResultBase
from api.user import user
from common.logger import loggerdef get_all_user_info():"""获取全部用户信息:return: 自定义的关键字返回结果 result"""result = ResultBase()  # 创建ResultBase对象，存储返回结果res = user.list_all_users()  # 调用API获取全部用户信息result.success = False  # 默认成功标志为Falseif res.json()["code"] == 0:  # 判断返回码是否为0result.success = True  # 如果返回码为0，表示成功else:result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])  # 错误信息result.msg = res.json()["msg"]  # 记录返回消息result.response = res  # 记录完整的响应return result  # 返回封装的结果def get_one_user_info(username):"""获取单个用户信息:param username:  用户名:return: 自定义的关键字返回结果 result"""result = ResultBase()  # 创建ResultBase对象res = user.list_one_user(username)  # 获取单个用户信息result.success = False  # 默认成功标志为Falseif res.json()["code"] == 0:result.success = True  # 成功时设置为Trueelse:result.error = "查询用户 ==>> 接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])  # 错误信息result.msg = res.json()["msg"]result.response = reslogger.info("查看单个用户 ==>> 返回结果 ==>> {}".format(result.response.text))  # 记录日志return resultdef register_user(username, password, telephone, sex="", address=""):"""注册用户信息:param username: 用户名:param password: 密码:param telephone: 手机号:param sex: 性别:param address: 联系地址:return: 自定义的关键字返回结果 result"""result = ResultBase()  # 创建ResultBase对象json_data = {  # 请求体内容"username": username,"password": password,"sex": sex,"telephone": telephone,"address": address}header = {  # 请求头"Content-Type": "application/json"}res = user.register(json=json_data, headers=header)  # 调用API进行注册result.success = False  # 默认失败if res.json()["code"] == 0:result.success = True  # 注册成功时标记为Trueelse:result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])  # 错误信息result.msg = res.json()["msg"]result.response = reslogger.info("注册用户 ==>> 返回结果 ==>> {}".format(result.response.text))  # 记录日志return resultdef login_user(username, password):"""登录用户:param username: 用户名:param password: 密码:return: 自定义的关键字返回结果 result"""result = ResultBase()  # 创建ResultBase对象payload = {  # 登录请求参数"username": username,"password": password}header = {"Content-Type": "application/x-www-form-urlencoded"}res = user.login(data=payload, headers=header)  # 调用API进行登录result.success = False  # 默认失败if res.json()["code"] == 0:result.success = True  # 登录成功时标记为True# result.token = res.json()["login_info"]["token"]  # 存储登录令牌else:result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])  # 错误信息result.msg = res.json()["msg"]result.response = reslogger.info("登录用户 ==>> 返回结果 ==>> {}".format(result.response.text))  # 记录日志return resultdef update_user(id, admin_user, new_password, new_telephone, token, new_sex="", new_address=""):"""根据用户ID，修改用户信息:param id: 用户ID:param admin_user: 当前操作的管理员用户:param new_password: 新密码:param new_telephone: 新手机号:param token: 当前管理员用户的token:param new_sex: 新性别:param new_address: 新联系地址:return: 自定义的关键字返回结果 result"""result = ResultBase()  # 创建ResultBase对象header = {"Content-Type": "application/json"}json_data = {  # 请求体内容"admin_user": admin_user,"password": new_password,"token": token,"sex": new_sex,"telephone": new_telephone,"address": new_address}res = user.update(id, json=json_data, headers=header)  # 调用API进行用户信息更新result.success = False  # 默认失败if res.json()["code"] == 0:result.success = True  # 更新成功时标记为Trueelse:result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])  # 错误信息result.msg = res.json()["msg"]result.response = reslogger.info("修改用户 ==>> 返回结果 ==>> {}".format(result.response.text))  # 记录日志return resultdef delete_user(username, admin_user, token):"""根据用户名，删除用户信息:param username: 用户名:param admin_user: 当前操作的管理员用户:param token: 当前管理员用户的token:return: 自定义的关键字返回结果 result"""result = ResultBase()  # 创建ResultBase对象json_data = {  # 请求体内容"admin_user": admin_user,"token": token,}header = {"Content-Type": "application/json"}res = user.delete(username, json=json_data, headers=header)  # 调用API进行用户删除result.success = False  # 默认失败if res.json()["code"] == 0:result.success = True  # 删除成功时标记为Trueelse:result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])  # 错误信息result.msg = res.json()["msg"]result.response = reslogger.info("删除用户 ==>> 返回结果 ==>> {}".format(result.response.text))  # 记录日志return result

业务api封装，供测试用例调用。

7. 测试报告展示

运行用例后，在项目根目录下，执行命令启动 allure 服务：

allure serve ./testcases/api_test/report