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/:接口封装层
python">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)
python">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级别日志
作用:
封装日志模块,记录测试执行过程。
使用方式:
python">from common.logger import loggerlogger.info("这是一条测试日志")
2.2 数据库操作(common/mysql_operate.py)
python">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 语句。
使用方式:
python">db = MysqlDb()
data = db.select_db("SELECT * FROM user")
2.3 读取配置文件(common/read_data.py)
python">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/:配置管理
python">[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 请求
python">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:封装接口返回结果
python">class ResultBase:"""结果基类,用于定义接口返回结果的标准结构。该类可以在后续扩展时添加通用的返回数据处理逻辑。"""pass
result_base.py 负责解析接口返回的 JSON 数据,并提供统一的访问方式,让测试用例更加简洁。
5.data/:测试数据管理
作用:
让测试用例数据分离,便于维护
支持数据驱动,自动化测试多个场景
6. operation/:业务封装层
python">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 服务:
python">allure serve ./testcases/api_test/report