想学习pytest,在gitee上找到了Pytest_For_Study这个项目,下载下来后debug运行了一遍,然后又跟着敲了一遍代码,运行报错,debug解决问题,将项目运行起来.在以上过程中,个人收益匪浅,最大的感触是:1.遇到问题简单看报错信息找不到原因,用debug模式排查问题;2.结合deepseek、kimi等AI工具查找问题答案是真的比翻阅多篇博客省事很多
贴出来Pytest_For_Study的源码下载地址: Pytest_For_Study: 一个基于 python语言 + pytest框架 的接口自动化项目Demo，非常适合接口自动化初学者，但是也需要一点代码基础哦，应用到实际项目上还需要改造，但是用于学习肯定够用啦！,下面学习分析下该项目
Pytest_For_Study项目没有调用数据库进行增删改查,而是写了一个mock(API_Mock_By_Flask包),模拟启动项目.然后运用API、Services、testcases、Tools思想,运行pytest框架进行自动化测试: API_Mock_By_Flask包模拟项目启动,API提供基础接口调用,API包内封装send_requests.py,便于services调用;testcases调用services,最后触发run_allure.py或run_pywebreport.py进行自动化断言
API_Mock_By_Flask包
API_Mock_By_Flask包含app.py,app.py代码实现了模拟项目启动
from flask import Flask,jsonify
app = Flask(__name__)
@app.route('/api/v1/addUser', methods=['POST'])
def addUser():
return jsonify({
"GatewayStatus": 1,
"GatewayMessage": "成功",
"DetailedStatus": 1,
"DetailedMessage": "新增成功",
"Data": {
"id": 1,
"name": "张三",
"sex": "男",
"age": 18,
"loginId": "test"
}
})
@app.route('/api/v1/getUserInfo', methods=['GET']) # 括号里写的是访问时的路径地址和请求方式
def getUSerInfo():
return jsonify({
"GatewayStatus": 1,
"GatewayMessage": "成功",
"DetailedStatus": 1,
"DetailedMessage": "查询成功",
"Data":{
"id": 1,
"name": "张三",
"sex":"男",
"age": "18",
"loginId": "test"
}
})
@app.route('/api/v1/deleteUser', methods=['DELETE'])
def deleteUser():
return jsonify({
"GatewayStatus": 1,
"GatewayMessage": "成功",
"DetailedStatus": 1,
"DetailedMessage": "删除成功"
})
@app.route('/api/v1/signLogin', methods=['POST']) # 括号里面写的是访问时的路径地址和请求方式
def userLogin():
return jsonify({
"GatewayStatus": 1,
"GatewayMessage": "成功",
"DetailedStatus": 1,
"DetailedMessage": "登录成功",
"Data": {
"token": "token2022",
"id": 1,
"name": "张三",
"sex": "男",
"age": 18,
"loginId": "test"
}
})
@app.route('/api/v1/RefreshToken', methods=['POST']) # 括号里写的是访问时的路径地址和请求方式
def refreshToken():
return jsonify({
"GatewayStatus": 1,
"GatewayMessage": "成功",
"DetailedStatus": 1,
"DetailedMessage": "刷新成功",
"Data": {
"token": "token20211012",
"id": 1,
"name": "张三",
"sex": "男",
"age": 18,
"loginId": "test"
}
})
@app.route('/api/v1/signout',methods=['POST']) # 括号里写的是访问时的路径地址和请求方式
def signOut():
return jsonify({
"GatewayStatus": 1,
"GatewayMessage": "成功",
"DetailedStatus": 1,
"DetailedMessage": "用户退出"
})
if __name__ == '__main__':
app.run(host='127.0.0.1')以上代码解析:
1.路由与请求方法
每个API使用@app.route定义路径和请求方法(POST/GET/DELETE)
示例:
@app.route('/api/v1/addUser', methods=['POST'])2.返回格式
同一返回JSON格式,包含状态码、消息和数据(GatewayStatus, DetailedMessage, Data等)
3.功能列表
addUser: 模拟新增用户(POST)
getUserInfo: 模拟查询用户(GET)
deleteUser: 模拟删除用户(DELETE)
signLogin: 模拟登录并返回Token(POST)RefreshToken: 模拟刷新Token(POST)
signout: 模拟用户退出(POST)
Tools包
Tools包将一些功能封装,提供接口,供pytest自动化运行时使用
Tools包含data_modify.py和log_output.py两个文件
先看data_modify.py
import os
import allure
import json
def data_replace(filepath, old_params, new_params):
"""
替换文件中的指定字符串（旧文案替换为新文案）
Args:
filepath (str): 文件路径
old_params (str): 需要被替换的旧字符串
new_params (str): 替换后的新字符串
Note:
- 以读写模式打开文件（'r+'）
- 先读取内容，清空文件后再逐行替换写入
- 确保文件编码为 UTF-8（避免中文乱码）
"""
with open(filepath, 'r+', encoding="utf-8") as f:
all_the_lines = f.readlines()
f.seek(0) # 移动指针到文件开头
f.truncate() # 清空文件内容
# 循环遍历每一行，替换字符串后重新写入
for line in all_the_lines:
line = line.replace(old_params, new_params)
f.write(line)
# 关闭文件
f.close()
def allure_step(step: str, var: str) -> None:
"""
添加带有附件（JSON 格式）的 Allure 测试步骤
Args:
step (str): 步骤名称（显示在报告中）
var (str): 需要附加的数据（自动转为 JSON 格式）
Note:
- 使用 allure.attach 附加 JSON 格式数据
- ensure_ascii=False 允许显示非 ASCII 字符（如中文）
- indent=4 美化 JSON 缩进
"""
with allure.step(step):
allure.attach(
json.dumps(
var,
ensure_ascii=False,
indent=4
),
step,
allure.attachment_type.JSON
)
def allure_step_no(step: str):
"""
添加无附件的 Allure 测试步骤（仅记录操作步骤）
Args:
step (str): 步骤名称（显示在报告中）
Note:
- 单纯用于标记测试步骤，不附加任何数据
"""
with allure.step(step):
pass以上代码,data_replace函数把文件里的旧自负全局替换为新字符串并保存;allure_step函数是记录报告步骤+JSON附件;allure_step_no函数是记录报告步骤,无附件
接下来看log_output.py文件
# 封装的log日志存储及输出方法
import logging
import os
def get_logger(name):
logger = logging.getLogger(name)
logger.setLevel(10) # 设置总日志等级
format = logging.Formatter(fmt='%(message)s') # 日志格式
cli_handler = logging.StreamHandler() # 输出到屏幕的日志处理器
log_result = os.path.join(os.path.dirname(__file__), r'../Testcases_Running_Log/case_run.log')
file_handler = logging.FileHandler(filename=log_result, mode='w', encoding='utf-8') # 输出到文件的日志处理器,mode默认是‘a’,即添加到文件末尾,‘w’是覆盖写入
cli_handler.setFormatter(format) # 设置屏幕日志格式
file_handler.setFormatter(format) # 设置文件日志格式
cli_handler.setLevel(logging.INFO) # 设置屏幕日志等级,可以大于日志记录器设置的总日志等级
logger.handlers.clear() # 清空已有处理器,避免继承了其他logger的已有处理器
logger.addHandler(file_handler) # 将文件日志处理器添加到logger
return logger以上代码,Logger是日志“总闸”,决定哪些级别的日志可以往下传;Handler是日志“出口”,决定写到哪(控制台、文件、邮件...);Formatter是日志“排版”,决定每条日志长什么样;Level是日志“门槛”:
DEBUG < INFO < WARNING < ERROR < CRITICAL
Api包
现在正式进入pytest自动化代码逻辑的范畴,Api包含send_requests.py,其代码如下
import requests
from Tools.log_output import get_logger
from Tools.data_modify import allure_step,allure_step_no
logger = get_logger('logger')
class BaseRequest(object):
session = None
@classmethod
def get_session(cls):
'''
单例模式保证测试过程中使用的都是一个session对象
:return:
'''
if cls.session is None:
cls.session = requests.Session()
return cls.session
@classmethod
def send_request(cls,url,method,parametric_key,headers=None,data=None,file=None):
'''处理case数据,转换成可用数据发送请求
:param case: 读取出来的每一行用例内容,可进行解包
return: 响应结果, 预期结果
'''
# 发送请求
response = cls.send_api(url, method, parametric_key, headers, data, file)
return response
@classmethod
def send_api(cls, url, method, parametric_key,headers=None,data=None,file=None) -> dict:
session = cls.get_session()
if parametric_key == 'params':
res = session.request(
method= method,
url = url,
params=data,
headers=headers
)
elif parametric_key == 'data':
res = session.request(
method=method,
url=url,
data=data,
files=file,
headers=headers
)
elif parametric_key == 'json':
res = session.request(
method=method,
url=url,
json=data,
files=file,
headers=headers
)
else:
raise ValueError(
'可选关键字为params,json,data')
response = res.json()
logger.info("\n\n*********** 用例执行记录 ****************")
logger.info(f'[{method}] [{res.status_code}] : URL >>> {url }')
logger.info(f"[请求Header ] : {headers}")
logger.info(f"[请求Body ] : {data}")
logger.info(f"[请求file ] : {file}")
logger.info(f"[请求返回 ] : {response}\n")
allure_step_no(f'响应耗时(s): {res.elapsed.total_seconds()}')
allure_step('请求Body: ', data)
allure_step_no(f"请求Header: {headers}")
allure_step('响应结果: ', response)
return response以上代码中,BaseRequest纯工具类(无实例),所有方法都是@classmethod,方便全局直接调用;session是类变量,单例模式的requests.Session对象,复用TCP链接,提升性能;get_logger('logger')从前面封装的日志工具里拿一个logger,所有请求都会写日志;allure_step*:把关键信息挂到Allure报告步骤里,方便CI/CD后查看
Services包
Services包含__init__.py、user_managment包,user_managment包中包含__init__.py和user_info_operation.
Services中的__init__.py代码展示
from . import user_managment
env = "DEV"
if env == "DEV":
ip_port = 'http://127.0.0.1:5000'
elif env == "TEST":
ip_port = 'http://www.4399.com'
else:
ip_port = 'http://www.baidu.com'根据当前环境变量 env 的值，把对应的 “服务器地址”（ip_port）设为 127.0.0.1:5000
user_managment中__init__.py代码展示
from . import user_info_operationuser_managment本身也是一个子包,from . import user_info_operation这行代码是把同级模块user_info_operation拉进来,绑定到
user_managment.user_info_operation,方便后续testcases引用
user_managment.user_info_operation.py代码展示
from Api.send_requests import BaseRequest
import Services
def add_user(user_name=None,sex=None,login_id=None, **kwargs):
'''
输入用户信息及登录账号,创建新用户
:param user_name: 用户姓名
:param sex: 用户性别
:param login_id: 用户登录ID
:param kwargs:
:return: 返回创建用户信息及创建状态
'''
url = Services.ip_port +'/api/v1/addUser'
method = 'POST'
parametric_key = 'json'
headers = {'Content-Type': 'application/json'}
data = {
'username': user_name,
'sex': sex,
'loginId': login_id
}
return BaseRequest.send_request(method=method,url=url,parametric_key=parametric_key,data=data,headers=headers)
def get_user_info(id=None,**kwargs):
'''
根据用户ID,查询用户信息
:param id: 数据库存储的用户ID
:param kwargs:
:return: 响应返回用户的详细信息
'''
url = Services.ip_port + '/api/v1/getUserInfo'
method = 'GET'
parametric_key = 'params'
headers = {'Content-Type': 'application/json'}
data = {
'id': id
}
return BaseRequest.send_request(method=method, url=url,parametric_key=parametric_key,data=data,headers=headers)
def delete_user(id=None,**kwargs):
'''
根据用户ID,查询用户信息
:param id: 数据库存储的用户ID
:param kwargs:
:return: 用户调用删除接口返回的操作状态
'''
url = Services.ip_port + '/api/v1/deleteUser'
method = 'DELETE'
parametric_key = 'json'
headers = {'Content-type': 'application/json'}
data = {
'id': id
}
return BaseRequest.send_request(method=method, url=url, parametric_key=parametric_key,data=data,headers=headers)
def user_sign_login(login_id=None,password=None,**kwargs):
'''
用户使用账号+密码进行登录
:param login_id: 用户的登录ID
:param password: 用户的密码
:param kwargs:
:return: 登录成功后返回用户数据及token信息
'''
url = Services.ip_port + '/api/v1/signLogin'
method = 'POST'
parametric_key = 'json'
headers = {'Content-Type': 'application/json'}
data = {
'loginId': login_id,
'passWord': password
}
return BaseRequest.send_request(method=method,url=url,parametric_key=parametric_key,data=data,headers=headers)
def user_token_refresh(token=None,**kwargs):
'''
使用token进行有效期延长
:param token: 用户登录成功后的token信息
:param kwargs:
:return: 刷新成功后返回用户数据及token信息
'''
url = Services.ip_port + '/api/v1/RefreshToken'
method = 'POST'
parametric_key = 'json'
headers = {'Content-Type': 'application/json'}
data = {
'token': token
}
return BaseRequest.send_request(method=method,url=url,parametric_key=parametric_key,data=data,headers=headers)
def user_signout(token=None,**kwargs):
'''
用户退出登录
:param token: 用户登录成功后的token信息
:param kwargs:
:return: 重置用户token以失效,返回退出状态
'''
url = Services.ip_port + '/api/v1/signout'
method = 'POST'
parametric_key = 'json'
headers = {'Content-Type': 'application/json'}
data = {
'token': token
}
return BaseRequest.send_request(method=method, url=url,parametric_key=parametric_key,data =data, headers=headers)以上代码是用户业务接口封装层:把6个最常用的用户操作(增、查、删、登录、续签、退出)整理成6个Python函数,内部统一调用BaseRequest发送HTTP请求,并自动拼接URL、方法、参数、Header.至此,测试用例里只需关心入参和断言,网络细节、日志、报告已全部被这一层屏蔽
testcases
testcases.test_user_operation.py代码如下:
import time
import allure
from Services import user_managment
from Tools.log_output import get_logger
import pytest
'''
@allure.severity(): severity_level 枚举
blocker: 阻塞缺陷(功能未实现,无法下一步)
critical: 严重缺陷(功能点缺失)
normal: 一般缺陷(边界情况,格式错误)
minor: 次要缺陷(界面错误与ui需求不符)
trivial: 轻微缺陷(必须项无提示,或者提示不规范)
'''
logger = get_logger('logger')
# 这里的fixture方法均为数据构建或前置条件
@pytest.fixture(scope="function")
def user_id():
new_user_info = dict({
'user_name': '张三',
'sex': '男',
'login_id': 'test1'
})
res = user_managment.user_info_operation.add_user(**new_user_info)
assert res["DetailedMessage"] == "新增成功"
user_id = res['Data']['id']
#yield 替代return,返回新增用户返回的用户ID
yield user_id
# 当user_id使用完毕后,调用删除用户接口,防止构建大量测试垃圾数据
res = user_managment.user_info_operation.delete_user(id=user_id)
assert res["DetailedMessage"] == "删除成功"
# 这里的fixture防伪均为数据构建或前置条件
@pytest.fixture(scope="function")
# 为了能够登录成功,需要以来用户新增成功
def token(user_id):
user_info = dict({
'login_id': 'test1',
'pass_word': '123456'
})
# 调用用户登录接口
res = user_managment.user_info_operation.user_sign_login(**user_info)
assert res["DetailedMessage"] == "登录成功"
token =res["Data"]["token"]
return token
@allure.feature("服务名称:用户管理服务")
@allure.story("用例模块: 用户增删改查相关模块")
class Test_User_Operation:
@allure.title("用例标题: 通过正确的用户ID,查询用户信息")
@allure.severity(allure.severity_level.BLOCKER)
@allure.description("用例描述:传入正确的用户ID,调用用户信息查询接口,断言响应结果")
def test_001_query_user_info(self,user_id):
'''查询用户信息'''
res = user_managment.user_info_operation.get_user_info(id=user_id)
time.sleep(0.5) # 为了看请求时间
assert res["DetailedMessage"] == "查询成功"
assert len(res["Data"]) >= 1
@allure.title("用例标题:刷新用户Token有效时常")
@allure.severity(allure.severity_level.CRITICAL)
@allure.description("用例描述: 传入用户的Token,调用有效期刷新接口,断言响应结果")
def test_002_user_token_refresh(self,token):
'''刷新用户token有效期'''
time.sleep(0.5) # 为了看请求时间
res = user_managment.user_info_operation.user_token_refresh(token=token)
assert res["DetailedMessage"] == "刷新成功"
assert res["GatewayMessage"] == "成功"
@allure.title("用例标题: 用户账号登出")
@allure.severity(allure.severity_level.NORMAL)
@allure.description("重置用户token,调用用户登出查询接口,断言相应结果")
def test_003_user_logout(self,token):
'''用户登出,将token有效期过期'''
time.sleep(0.5) # 为了看请求时间
res = user_managment.user_info_operation.user_signout(token=token)
assert res["DetailedMessage"] == "用户退出"
assert res["GatewayMessage"] == "成功"以上代码相当于pytest+allure的用户管理接口自动化测试文件,它用fixture做前置数据(先创用户->再删用户),然后对查询、刷新Token、登出三个场景分别写用例,并自动生成Allure报告
run_allure.py
接下来看启动pytest自动化的代码
import os
import pytest
from Tools.log_output import get_logger
logger = get_logger('logger')
def run():
logger.info("""
__ _ _ __ (_) / \\ _ _| |_ __|_ _|__ ___| |_
/ _` | '_ \\| | / _ \\| | | | __/ _ \\| |/ _ \\/ __| __|
| (_| | |_) | |/ ___ \\ |_| | || (_) | | __/\\__ \\ |_
\\__,_| .__/|_/_/ \\_\\__,_|\\__\\___/|_|\\___||___/\\__|
|_|
Starting ... ... ... ...
""")
'''
1脚本执行方法一:
运行Pytest命令,执行脚本[可自定义测试文件/测试文件夹]
'''
pytest.main(['-sv','./testcases','--alluredir','./Allure_Result'])
'''
2脚本执行方法二:
使用cmd命令执行脚本
'''
# 获取当前文件所在文件夹路径,也就是项目文件夹
project_path = os.getcwd()
#测试报告json数据存储路径
result_path = os.path.join(project_path,'Allure_Result')
# allure测试保存存储路径
report_path = os.path.join(project_path, 'Allure_Report')
# 调用 cmd 命令, 将result_path的json数据转化为Allure报告并存储在report_path目录下
cmd_command = 'allure generate ' + result_path + ' -o ' + report_path + ' --clean'
logger.info("命令行生成报告命令: " + cmd_command)
os.system(cmd_command)
# 自动打开生成的 Allure 测试报告
open_report_command = 'allure open -h 127.0.0.1 -p 6789 ./Allure_Report'
os.system(open_report_command)
logger.info(f'报告已生成')
if __name__ == '__main__':
'''
run脚本执行注意事项:
1、执行方法前,先确保API_Mock_By_Flask文件下的app.py文件已经run起来了哦
2.生成的测试报告html文件路径: Allure_Report文件下的index.html
'''
run()
以上两种方法都可正确运行代码
run_pywebreport.py
也可通过pywebreport启动工程,页面定制化,内容更简洁
import os
import pytest
from Tools.log_output import get_logger
from Tools.data_modify import data_replace
logger = get_logger('logger')
def run():
logger.info("""
__ _ _ __ (_) / \\ _ _| |_ __|_ _|__ ___| |_
/ _` | '_ \\| | / _ \\| | | | __/ _ \\| |/ _ \\/ __| __|
| (_| | |_) | |/ ___ \\ |_| | || (_) | | __/\\__ \\ |_
\\__,_| .__/|_/_/ \\_\\__,_|\\__\\___/|_|\\___||___/\\__|
|_|
Starting ... ... ...
""")
# 脚本执行方法一：
args = ['.', '-q', '--report', 'Pywebreport_Result_and_Report/report.html']
pytest.main(args)
report_html_path = os.path.join(os.path.dirname(__file__), r"Pywebreport_Result_and_Report/report.html")
data_replace(filepath=report_html_path,old_params="PyWebReport</div>",new_params="PythonKimo</div>")
data_replace(filepath=report_html_path,old_params='id="dashboard" class="tabcontent" style="display: none', new_params='id="dashboard" class="tabcontent" style="display: block')
if __name__ == '__main__':
'''
run 脚本执行注意事项：
1、执行方法前，先确保 API_Mock_By_Flask 文件下的 app.py 文件已经 run 起来了哦~
2、生成的测试报告html文件路径： Pywebreport_Result_and_Report 文件下的 report.html
'''
run()至此,该项目梳理完成了,学习一遍就有一次收获,后续还需要继续加强学习