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类，创建日志记录器
"""
# 定义日志文件的完整路径，文件名为当前日期.log
self.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()
# 设置控制台日志级别为DEBUG
self.console.setLevel(logging.DEBUG)
# 设置文件日志级别为DEBUG
self.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().logger
if __name__ == '__main__':
logger.info("---测试开始---")# 记录INFO级别日志
logger.debug("---测试结束---")# 记录DEBUG级别日志
作用：
封装日志模块，记录测试实验过程。
使用方式：
from common.logger import logger
logger.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 data
def 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 optionstr
class 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 data
def 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 data
def 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/：设置管理
# 测试环境
api_root_url = http://127.0.0.1:5000
# 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根URL
self.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# 拼接完整的URL
headers = 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 logger
def get_all_user_info():
"""
获取全部用户信息
:return: 自定义的关键字返回结果 result
"""
result = ResultBase()# 创建ResultBase对象，存储返回结果
res = user.list_all_users()# 调用API获取全部用户信息
result.success = False# 默认成功标志为False
if res.json()["code"] == 0:# 判断返回码是否为0
result.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# 默认成功标志为False
if res.json()["code"] == 0:
result.success = True# 成功时设置为True
else:
result.error = "查询用户 ==>> 接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])# 错误信息
result.msg = res.json()["msg"]
result.response = res
logger.info("查看单个用户 ==>> 返回结果 ==>> {}".format(result.response.text))# 记录日志
return result
def 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# 注册成功时标记为True
else:
result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])# 错误信息
result.msg = res.json()["msg"]
result.response = res
logger.info("注册用户 ==>> 返回结果 ==>> {}".format(result.response.text))# 记录日志
return result
def 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 = res
logger.info("登录用户 ==>> 返回结果 ==>> {}".format(result.response.text))# 记录日志
return result
def 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# 更新成功时标记为True
else:
result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])# 错误信息
result.msg = res.json()["msg"]
result.response = res
logger.info("修改用户 ==>> 返回结果 ==>> {}".format(result.response.text))# 记录日志
return result
def 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# 删除成功时标记为True
else:
result.error = "接口返回码是 【 {} 】, 返回信息：{} ".format(res.json()["code"], res.json()["msg"])# 错误信息
result.msg = res.json()["msg"]
result.response = res
logger.info("删除用户 ==>> 返回结果 ==>> {}".format(result.response.text))# 记录日志
return result
业务api封装，供测试用例调用。
7. 测试报告展示
运行用例后，在项目根目录下，实验命令启动 allure 服务：
allure serve ./testcases/api_test/report
https://i-blog.csdnimg.cn/direct/7d5fb462b3c940d791723030cb748bd9.png
https://i-blog.csdnimg.cn/direct/756d739f1181421189540b734eebf95d.png
https://i-blog.csdnimg.cn/direct/9c2d95cbc6064af2928d07007aa9cff5.png
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]