Python 中的*args 与**kwargs
接收可变参数
- *args:会将传入的额外位置参数打包成一个元组(tuple)
- **kwargs:会将传入的额外关键字参数打包成一个字典(dict)
# 调用时,参数只需传递一个一个的值,自动会打包为元组 args
def my_tool(*args):
print(args)
():
(kwargs)
__name__ == :
my_tool(, , )
my_tools(name=, age=)
基于Python结合Requests库发送HTTP请求,利用Pytest作为测试执行器,Allure生成报告,以及YAML进行数据驱动的接口自动化测试框架搭建过程。涵盖关键字封装、全局配置管理、用例数据关联、断言逻辑(含数据库)、日志集成、Session复用及文件上传下载等功能,并提供了项目打包部署方案。
接收可变参数
# 调用时,参数只需传递一个一个的值,自动会打包为元组 args
def my_tool(*args):
print(args)
():
(kwargs)
__name__ == :
my_tool(, , )
my_tools(name=, age=)
解包
data = {'username': 'zs', 'password': '123'}
names = ['zs', 'ls', 'wu']
def func1(username, password):
print(username + password)
def func2(name1, name2, name3):
print(name1 + name2 + name3)
if __name__ == '__main__':
func1(**data) # zs123
func2(*names) # zslswu
需要先安装:pip install requests
import requests
url = 'http://127.0.0.1:5000/hello'
data = {'username': 'zs', 'password': 123}
# 发送 get 请求
# 请求方式一:requests.request(method, url, **kwargs)
ret = requests.request('get', url=url)
print(ret.json())
# 请求方式二:requests.get(url, params=None, **kwargs)
ret = requests.get(url=url, params=data)
print(ret.json())
ret = requests.request('post', url=url, json=data)
ret = requests.request('post', url=url, data=data)
print(ret.json())
ret = requests.post(url=url, json=data)
ret = requests.post(url=url, data=data)
print(ret.json())
# data 与 json 参数区别:
# json 要求传字典,json.loads 将字符串转为字典,content-type 为 application/json
# data 要求字符串,json.dumps 将字典转为字符串,content-type 不是 json 一般为 application/x-www-form-urlencoded
# data 可以传字典等格式数据,只是会自动编码为字符串,如果传的本身是字符串,则不进行编码
data 与 json 参数区别
想在 url 地址栏增加一些参数,可以指定 params 参数
params = {'page': 2, 'len': 10}
requests.post(url=url, json=data, params=params)
上传文件,传递 files 参数
# 上传文件
ret = requests.post(url=url, files={'image': open('a.jpg', 'rb')})
requests 发送请求是独立的,但是有时需要保存会话内容
session = requests.session()
# 登录
ret = session.post(url=url, json=data)
# 登录后操作
ret = session.post(url=url, json=data)
jsonpath 用于在 json 数据中定位提取数据 使用前需要先安装:pip install jsonpath
规则:
import jsonpath
jsonpath.jsonpath(response, '$.name')
jsonpath.jsonpath(response, '$..title')
jsonpath.jsonpath(response, '$.names[0]')
jsonpath.jsonpath(response, '$.names[*]')
jsonpath.jsonpath(response, '$.books[0,3]') # 返回 book 数组中 category 为 fiction 的元素
jsonpath.jsonpath(response, "$.store.book[?(@.category == 'fiction')]")
main 文件:程序入口
使用前安装:pip install pyyaml 格式:
base_config:
case_type: 'ApiCase'
case_name: 'login'
case_module: '教务系统'
author: 'hlk'
test_steps:
- 登录接口:
request_type: send_post
url: 'http://127.0.0.1'
params:
s: /hello
pageIndex: 2
data_type: json
request_data:
username: hlk
password: 123
import yaml
def read_yaml(file_path):
data = []
with open(file_path, 'r', encoding='utf-8') as file:
data.append(yaml.safe_load(file))
return data
关键字类中存放一些方法,如发送请求,解析响应等关键字方法
class MyRequest:
def __init__(self, request):
self.request = request
def send_post(self, **kwargs):
url = kwargs.get('url', None)
params = kwargs.get('params', None)
headers = kwargs.get('headers', None)
files = kwargs.get('files', None)
data = kwargs.get('data', None)
request_data = {'url': url, 'params': params, 'headers': headers, 'files': files}
if kwargs.get('data_type') == 'json':
request_data['json'] = data
if kwargs.get('data_type') == 'data':
request_data['data'] = data
try:
ret = self.request.request('post', **request_data)
except Exception as e:
print('请求异常:', e)
return ret
import pytest, requests
from api_frame.HAT.parse.read_yml import read_yaml
from api_frame.HAT.utils.my_request import MyRequest
class TestRunner:
case_info = read_yaml('../../test_cases/yml_cases/login.yml')
@pytest.mark.parametrize('case_info', case_info)
# case_info 要求是列表
def test_excute_case(self, case_info):
my_request = MyRequest(requests)
base_config = case_info.get('base_config', {}) # 基础配置,用于报告
test_steps = case_info.get('test_steps', []) # 测试步骤列表
for test_step in test_steps:
# step_name = list(test_step.keys())[0]
# step_value = list(test_step.values())[0]
(step_name, step_value), = test_step.items() # 注意逗号,解包一个包含一个元素的可迭代对象
request_type = step_value.get('request_type') # 发送请求
try:
request_func = my_request.__getattribute__(request_type) # 获取请求方法
except Exception as e:
print('my_request 中没有找到对应的方法', e)
ret = request_func(**step_value).json()
print(ret)
if __name__ == '__main__':
pytest.main(['-vs', __file__])
创建 main 文件,在 main 文件中执行用例生成 allure 报告 安装配置 allure:
import pytest, os
from allure_combine import combine_allure
# -vs 详细内容
# --capture=sys 系统配置 生成 stdout 附件
# --clean-alluredir 清空报告报告数据,保证每次生成报告都使用最新的数据
# --alluredir=allure-results 结果数据目录,生成的 原始测试结果文件(JSON 格式)
# ./core/test_runner.py 用例文件夹
pytest_args = ['-vs', '--capture=sys', '--clean-alluredir', '--alluredir=allure-results', './core/test_runner.py']
pytest.main(pytest_args)
# 生成测试报告
# allure-report 最终测试报告目录,根据 allure-results 中数据创建报告
os.system('allure generate allure-results -o allure-report -c')
# -o 指定目录 -c 先清空在生成
# 生成的报告只能在 pycharm 打开,需要安装 allure-combine
# npm install -g allure-combine
combine_allure('./allure-report')
base_config = case_info.get('base_config', {})
allure.dynamic.parameter('case_info', '') # 记录参数
allure.dynamic.feature(base_config.get('case_module')) # 用例所属模块
allure.dynamic.story(base_config.get('case_module_sec')) # 模块子功能
allure.dynamic.title(base_config.get('case_name')) # 用例标题
with allure.step(step_name): # 划分步骤,必须在函数中使用
request_type = step_value.get('request_type') # 发送请求
try:
request_func = my_request.__getattribute__(request_type) # 获取请求方法
except Exception as e:
print('my_request 中没有找到对应的方法', e)
ret = request_func(**step_value).json()
print(ret)
# 调用该函数时,自动创建 step
@allure.step('发送 post 请求')
def send_post(self, **kwargs):
url = kwargs.get('url', None)
params = kwargs.get('params', None)
headers = kwargs.get('headers', None)
files = kwargs.get('files', None)
data = kwargs.get('data', None)
需要安装:pip install tqdm 进度条库,终端输出工具 注意导包:from tqdm import tqdm
test_steps = case_info.get('test_steps', []) # 测试步骤列表
# total=len(test_steps):告诉 tqdm 总共有多少步(用于计算百分比)
# desc='开始执行':初始描述文字(显示在进度条左边)
# with ... as pbar:使用上下文管理器,确保进度条正确关闭
# pbar:进度条对象,后续可通过它动态更新内容
with tqdm(total=len(test_steps), desc='开始执行') as pbar:
for test_step in test_steps:
# step_name = list(test_step.keys())[0]
# step_value = list(test_step.values())[0]
(step_name, step_value), = test_step.items() # 注意逗号,解包一个包含一个元素的可迭代对象
pbar.set_description(f'{base_config.get("case_name")}-当前步骤:{step_name}') # 进度条描述,进度条左侧的文字
pbar.update(1) # 更新进度,进度条前进一格
为什么 tqdm 进度条会出现在 Allure 报告里?
一些全局配置如项目 url 地址可以单独存放在一个 yaml 文件中,创建 context.yml 保存这些数据
login_url: 'http://127.0.0.1:8080/login'
register_url: 'http://127.0.0.1:8080/register'
file_upload_url: 'http://127.0.0.1:8080/upload'
file_download_url: 'http://127.0.0.1:8080/download/a.png'
databases:
hlk_test_dev:
user: 'root'
database: 'hlk_test_dev'
password: '12345678'
host: 'localhost'
port: 3306
session_reuse: True
在用例 yaml 中使用这些配置,可以使用{{变量}}
test_steps:
- 添加接口:
request_type: send_post
url: '{{login_url}}'
data_type: json
request_data:
username: hlk
password: 123
token: '{{token}}'
多个 yaml 之间数据传递思路:先将配置数据保存到全局变量中,再将全局变量渲染到 yaml 用例中
class GlobalVar:
_dict = {}
def set_dict(self, dict):
# 先查找实例属性_dict,找不到再查找类属性_dict
# 如果使用赋值操作如 self._dict = {}, 则会创建一个实例属性_dict,不会修改到类属性
# 如果使用赋值操作 (self.attr = value),会优先在实例级别创建一个新的属性 attr,而不会修改类属性
self._dict.update(dict)
# GlobalVar._dict.update(dict)
# 添加@classmethod
def set_dict_by_kv(self, key, value):
self._dict[key] = value
def get_dict(self):
return self._dict
def get_dict_by_key(self, key):
return self._dict.get(key)
global_var = read_context('./test_cases/yaml_cases/context.yml')
# def read_context(file_path):
# with open(file_path, 'r', encoding='utf-8') as file:
# data = yaml.load(file, Loader=yaml.SafeLoader)
# return data
GlobalVar().set_dict(global_var) # 设置到全局变量中
使用 jinjia2 进行渲染,需要安装:pip install jinja2
from jinja2 import Template
# jinja2 的模版必须是字符串,且渲染后返回字符串
# 渲染 yml 文件,eval 将字符串转为原本字典
step_value = eval(Template(str(step_value)).render(GlobalVar().get_dict()))
yml 用例中添加前置及后置操作,此处是直接执行代码方式,可优化为调方法执行
base_config:
case_type: 'ApiCase'
case_name: 'login'
case_story: '登录模块'
case_module: '教务系统'
author: 'hlk'
setup_script:
- "print('前置执行 1')"
- "print('前置执行 2')"
teardown_script:
- "print('后置执行 1')"
- "print('后置执行 2')"
test_steps:
- 登录接口:
request_type: send_post
url: '{{login_url}}'
data_type: json
request_data:
username: hlk
password: 123
核心执行器中添加前置与后置脚本执行的代码:
# 前置脚本执行
setup_scripts = case_info.get('setup_script')
if setup_scripts:
for setup_script in setup_scripts:
# 执行任意代码
exec(setup_script, {'context': GlobalVar().get_dict()})
# 后置脚本执行
teardown_scripts = case_info.get('teardown_script')
if teardown_scripts:
for teardown_script in teardown_scripts:
# 执行任意代码
exec(teardown_script, {'context': GlobalVar().get_dict()})
exec 函数:动态执行代码 exec(表达式,全局变量作用域,局部变量作用域),如果指定了局部变量作用域,则表达式生成的变量将会存在这里,不会污染全局作用域变量
执行多个用例: 按照特定规则读取指定目录下的 yml 文件
# 读取文件夹下符合规则的 yaml 文件
def read_rule_yaml(file_dir):
case_infos = []
files = os.listdir(file_dir) # 列表推导式,yml 文件且是 test 结尾的
rule_files = [file for file in files if file.endswith('yml') and file.split('.')[0].endswith('test')]
rule_files.sort()
for rule_file in rule_files:
file = os.path.join(file_dir, rule_file) # 文件路径
with open(file, 'r', encoding='utf-8') as f:
case_info = yaml.safe_load(f)
case_infos.append(case_info)
return case_infos
核心执行器中替换读取用例的方法
case_info = read_rule_yaml('../test_cases/yaml_cases/') # 读取文件夹下满足规则的所有用例
多个用例关联起来:
每次请求可以将响应数据保存到全局变量中,发送请求的方法增加保存响应数据到全局变量:
response = self.request.request('post', **request_data) # 将响应保存到全局变量中
GlobalVar().set_dict({'response_data': response.json()})
yaml 用例文件增加提取步骤:
test_steps:
- 登录接口:
request_type: send_post
url: '{{login_url}}'
data_type: json
request_data:
username: hlk
password: 123
- 提取保存数据:
request_type: extract_data
extract_expression: '$..token'
extract_var: 'token'
@allure.step('提取保存数据')
def extract_data(self, **kwargs):
index = kwargs.get('extract_index', None)
if index == None:
# 提取下标默认为 0
index = 0
response = GlobalVar().get_dict_by_key('response_data')
extract_expression = kwargs.get('extract_expression', None)
extract_var = kwargs.get('extract_var', None)
extract_value = jsonpath.jsonpath(response, extract_expression)[index] # 从响应中提取数据
if not extract_value:
raise Exception(f'没有从响应中取到对应值{extract_expression}')
GlobalVar().set_dict_by_kv(extract_var, extract_value)
print(f'提取{extract_var}的值{extract_value}')
下一个用例执行时,从全局变量中取需要的字段,之前的核心执行器中已经支持从全局变量渲染用例文件,所以此处只需使用{{}}替换变量值
核心执行器渲染代码
step_value = eval(Template(str(step_value)).render(GlobalVar().get_dict()))
yml 文件
- 删除接口:
request_type: send_post
url: '{{register_url}}'
data_type: json
request_data:
username: hlk
password: 123
token: '{{token}}'
yml 用例中添加断言内容
- 断言数据:
request_type: validate_data
validate:
- ==:
actual: '{{msg}}' # 实际值已经在上述步骤中添加到全局变量中了
expect: good
assert_fail_reason: '实际与预期不相等'
- in:
actual: '{{msg}}'
expect: 'goods'
关键字类中增加断言方法
@allure.step('断言数据')
def validate_data(self, **kwargs):
comparators = {
'==': lambda x, y: x == y,
'!=': lambda x, y: x != y,
'>=': lambda x, y: x >= y,
'<=': lambda x, y: x <= y,
'in': lambda x, y: x in y,
'not in': lambda x, y: x not in y,
}
validates = kwargs.get('validate')
for validate in validates:
(comparator, validate_data), = validate.items()
if comparator not in comparators:
raise Exception(f'{comparator}比较符不在比较器中')
expect = validate_data.get('expect')
actual = validate_data.get('actual')
assert_fail_reason = validate_data.get('assert_fail_reason', None)
if not comparators[comparator](expect, actual):
if assert_fail_reason:
raise AssertionError(assert_fail_reason)
else:
raise AssertionError(f'{actual} 与 {expect} 不相等')
需要使用 deepdiff,安装:pip install deepdiff 可以设置忽略对比的字段,忽略大小写,忽略列表元素的顺序等
yml 文件增加批量断言步骤
- 断言全部数据:
request_type: validate_all_data
expect:
code: 20
msg: 'goosd'
data:
token: 'esjdksaos'
name: 'zs'
uid: '44'
auths: ['prod', 'dev']
filter_fields: ["root['data']['uid']", "root['data']['token']"]
关键字类中增加批量断言方法
@allure.step('断言全部响应数据')
def validate_all_data(self, **kwargs):
try:
actual_response = GlobalVar.get_dict_by_key('response_data') # 上次接口请求已经将响应放到全局变量里
expect_response = kwargs.get('expect')
filter_fields = kwargs.get('filter_fields', [])
ignore_order = kwargs.get('ignore_order', True) # 默认忽略排序
ignore_case = kwargs.get('ignore_case', True) # 默认忽略大小写
data = {
'exclude_paths': filter_fields,
'ignore_order': ignore_order,
'ignore_string_case': ignore_case
}
diff = DeepDiff(actual_response, expect_response, **data)
except Exception as e:
assertFalse, f'批量断言失败:{e}'
# 如果 diff 成功则返回空列表,如果 diff 存在差异则返回有差异的信息
# 所以希望 diff 为空时校验通过,diff 不为空时校验失败,所以需要加 not
assert not diff, f'批量断言失败{diff}'
操作数据库过程
使用前需安装:pip install pymysql
import pymysql
from pymysql import cursors
# 数据库链接
connect = pymysql.connect(
user='root',
database='hlk_test_dev',
password='12345678',
host='localhost',
port=3306,
cursorclass=cursors.DictCursor, # 字典形式展示数据
charset='utf8'
)
# 生成游标
cursor = connect.cursor()
# 游标执行 sql
sql = 'select * from user'
cursor.execute(sql)
# 游标获取结果
ret = cursor.fetchall()
print(ret)
# 关闭游标,关闭数据库
cursor.close()
connect.close()
yml 用例中先提取查询的数据库结果,再进行数据库断言
- 提取数据库数据:
request_type: extract_database
database_name: hlk_test_dev
sql: select * from user
extract_var: [username, password]
- 断言数据库:
request_type: validate_data
validate:
- in:
actual: '''{{username}}'''
expect: 'zs'
assert_fail_reason: '实际与预期不相等'
数据库配置放在全局配置文件 context.yml 中
databases:
hlk_test_dev:
user: 'root'
database: 'hlk_test_dev'
password: '12345678'
host: 'localhost'
port: 3306
提取数据库数据关键字方法
@allure.step('提取数据库数据')
def extract_database(self, **kwargs):
database_name = kwargs.get('database_name')
db_config = GlobalVar().get_dict_by_key('databases').get(database_name)
config = {'cursorclass': cursors.DictCursor, 'charset': 'utf8'}
config.update(db_config)
connect = pymysql.connect(**config)
# 生成游标
cursor = connect.cursor()
# 游标执行 sql
cursor.execute(kwargs.get('sql'))
# 游标获取结果
ret = cursor.fetchall()
# [{'id': 1, 'username': 'zs', 'password': '123'}]
# 关闭游标,关闭数据库
cursor.close()
connect.close()
# 提取保存数据
result = {}
vars = kwargs.get('extract_var') # 保存的变量名 [username, password]
# 查询如果是多个结果则放在数组中
for var in vars:
if len(ret) <= 1:
result[var] = ''
else:
result[var] = []
for i in ret: # [{'id': 1, 'username': 'zs', 'password': '123'}]
for key, value in i.items():
if key == var:
if len(ret) <= 1:
result[var] = value
else:
result[var].append(value)
# result = {'username':[], 'password':[]}
GlobalVar().set_dict(result) # 保存到全局变量
修改 yaml 用例文件:
test_steps:
- 登录接口:
request_type: send_post
url: '{{login_url}}'
data_type: json
request_data:
username: '{{username}}'
password: '{{password}}'
- 提取响应数据 msg:
request_type: extract_data
extract_expression: '$..msg'
extract_var: 'msg'
- 断言数据:
request_type: validate_data
validate:
- ==:
actual: '{{msg}}'
expect: '{{expect}}'
assert_fail_reason: '实际与预期不相等'
数据驱动
ddt:
- title: '正确用户名'
username: hlk
password: 123
expect: 'login success'
- title: '错误用户名'
username: zs
password: 456
expect: 'login fail'
解析数据驱动思路:
解析数据驱动:
def ddt_yaml_parse(file_dir):
case_infos = read_rule_yaml(file_dir)
new_case_infos = []
for case_info in case_infos:
ddts = case_info.get('ddt') # 数据驱动
if ddts:
case_info.pop('ddt') # 存在数据驱动时,遍历每个数据驱动,组装每一条新的 case
for ddt in ddts:
# 注意需要深拷贝,否则对同一变量修改,之前 new_case_infos 添加的也就同步被修改了
case_info = copy.deepcopy(case_info)
case_info.update({'ddt': ddt})
new_case_infos.append((case_info))
else:
new_case_infos.append(case_info) # 无数据驱动时直接添加到 case 数组中
return new_case_infos
核心执行器执行时会将全局变量渲染到用例文件 core 核心执行器中,增加将数据驱动保存到全局变量:
# 在模版渲染前,将 ddt 保存到全局变量中
GlobalVar().set_dict(case_info.get('ddt'))
安装:pip install loguru 基本使用:
from loguru import logger
# 保存到文件中
logger.add('log.log',
rotation='100 MB', # 单个日志文件大小
retention='10 days') # 保存时间
# 日志级别:debug<info<warning<error
logger.debug('调试级别')
logger.info('程序正常运行')
logger.warning('警告信息')
logger.error('错误信息')
try:
a = 1/0
logger.info('run...')
except Exception as e:
logger.error(e)
time_str = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
logger.configure(handlers=[
{"sink": sys.stdout, "level": 'DEBUG'},
{"sink": sys.stderr, "level": 'DEBUG'},
{"sink": f'./log/log_{time_str}.log', "level": 'DEBUG'}
])
程序入口 main 文件中新增日志配置:
# 配置日志
if not os.path.exists('./log'):
os.mkdir('log')
time_str = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
logger.configure(handlers=[
{"sink": sys.stdout, "level": 'DEBUG'},
{"sink": sys.stderr, "level": 'DEBUG'},
{"sink": f'./log/log_{time_str}.log', "level": 'DEBUG'}
])
步骤与日志关联,日志信息放在每个 allure 测试步骤下面:
import io
import allure
from loguru import logger
# 日志管理器
class AllureStepLogger():
def __init__(self):
# io 流从内存中读取数据,区别 open 是从文件读取数据
# StringIO 读取文本,BytesIO 读取二进制
self.log_buffer = io.StringIO()
# log_buffer 存储读取到内存的日志信息
self.sink_id = None
# 上下文管理器,使用到 with 语句,会调__enter__方法
def __enter__(self):
self.sink_id = logger.add(self.log_buffer, level='DEBUG')
# 退出 with 语句时,会调__exit__方法
def __exit__(self, exc_type, exc_val, exc_tb):
logger.remove(self.sink_id)
log = self.log_buffer.getvalue() # 获取 io 流读取到内存的日志信息
# 添加到 allure 中
if log.strip(): # strip 去除首和尾指定字符,默认空格
allure.attach(
body=log, # 附件内容
name='步骤日志', # 附件名称
attachment_type=allure.attachment_type.TEXT # 附件类型,JSON,ZIP,TEXT
# , # 附件扩展名,不提供则根据附件类型推测
)
# 向 allure 报告中添加附件
self.log_buffer.close()
# 关联测试步骤与报告
from contextlib import contextmanager
@contextmanager # 装饰器,该函数可以使用 with 语句
def allure_step_with_log(step_name):
with allure.step(step_name):
with AllureStepLogger() as collector:
# yield 之前代码相当于__enter__,yield 之后相当于__exit__,yield 后的值返回给调用方
# @contextmanager 必须和 yield 同时使用
# yield 暂停函数的执行,并返回一个值给调用方,将控制权交回调用方
# 这里每次使用 with 调用时都会创建新对象,与之前调用无关
yield collector
with allure_step_with_log(step_name): # with allure.step(step_name): # 报告右侧显示的步骤
日志器调用顺序梳理:
在 context.yml 中可以配置 session 是否复用
session_reuse: True
根据配置选择初始化方式:
# 用于判断是否 session 复用,决定使用何种方式进行实例化关键字类
global_session = None
class ApiCaseContext:
def __init__(self):
self.request = None
self.my_request = None
def init_kwargs_func(self):
global global_session
session_reuse = GlobalContext().get_dict_by_key('session_reuse') # 从全局变量拿配置
if session_reuse is not None and session_reuse == True:
if global_session == None:
global_session = requests.session()
self.request = global_session
else:
self.request = requests
self.my_request = MyRequest(self.request)
return self.my_request
核心执行器中替换关键字类的实例化方式:
# my_request = MyRequest(requests)
my_request = ApiCaseContext().init_kwargs_func()
yml 用例中修改请求数据类型为 files
test_steps:
- 文件上传接口:
request_type: send_post
url: '{{file_upload_url}}'
data_type: files
request_data:
img: 'a.png' # 文件路径
发送请求方法增加判断操作文件修改请求参数:
if data_type == 'files':
if isinstance(data, dict):
files = {}
for key, file_path in data.items():
files[key] = open(file_path, 'rb')
request_data['files'] = files
else:
file = open(data, 'rb')
request_data['files'] = {'files': file}
yml 用例增加下载步骤:
- 文件下载接口:
request_type: send_get
url: '{{file_download_url}}'
save_path: '/Users/user/Desktop/hlk_test_dev/requests_test/HAT/utils/download/b.png' # 保存文件路径
stream: True # 流式下载 (可不设置)
chunk_size: 1024 # 块大小 (可不设置)
发送 get 请求方法增加判断下载文件的操作:
@allure.step('发送 get 请求')
def send_get(self, **kwargs):
url = kwargs.get('url')
params = kwargs.get('params')
data_type = kwargs.get('data_type')
data = kwargs.get('request_data')
save_path = kwargs.get('save_path')
stream = kwargs.get('stream', False) # 流式下载
chunk_size = kwargs.get('chunk_size', 1024) # 块大小
request_data = {'url': url, 'params': params, 'stream': stream if save_path else False,}
if data_type == 'json':
request_data['json'] = data
if data_type == 'data':
request_data['data'] = data
try:
response = self.request.request('get', **request_data)
logger.debug('发送请求 success')
if save_path:
# 确保目录已存在,exist_ok=True:如果已存在不会抛异常
os.makedirs(os.path.dirname(save_path), exist_ok=True)
if stream: # 大文件保存
with open(save_path, 'wb') as file:
# response.iter_content 以固定大小(chunk_size)分块读取响应内容
for chunk in response.iter_content(chunk_size=chunk_size):
file.write(chunk)
else: # 正常文件保存
with open(save_path, 'wb') as file:
file.write(response.content) # response.content 获取二进制
except Exception as e:
print('请求错误:', e)
return response
给 pytest 添加两个自定义执行参数,case(用例目录),type(用例格式 yml 或 excel), pytest 原本不支持该参数,现在扩展支持让 pytest 支持 case 目录参数,type 用例格式参数
修改执行入口 main 文件,添加自定义执行参数与插件
pytest_args = ['-vs', '--capture=sys', '--clean-alluredir', '--alluredir=allure-results', './core/core.py', # 核心执行器的目录
'--case=../test_cases/yaml_cases_01', # 自定义执行 case 目录参数
'--type=yml' # 自定义执行用例格式 yml,excel 参数]
# plugins,插件列表,通常传入类实例或模块
# pytest 运行时,会自动扫描所有加载的插件(包括 conftest.py 和通过 plugins 参数传入的插件)
pytest.main(pytest_args, plugins=[CasePlugin()])
创建自定义插件类: 将钩子函数定义在插件中,pytest 运行时会扫描加载的插件和测试代码,寻找这些钩子函数,匹配到就会在合适的时机调用它,这里使用到的钩子函数:
import pytest
from requests_test.HAT.parse.yaml_parse import parse_pytest_parameter # 钩子函数可以放在 conftest 文件,pytest.ini 文件,插件
# 将钩子函数定义在自定义插件中
class CasePlugin:
# 固定名称 pytest_addoption,pytest 钩子函数,用于向 pytest 的命令行添加自定义参数
def pytest_addoption(self, parser):
parser.addoption('--case', # 参数名
action='store', # 表示需要存储一个值
default='', # 默认值
help='用例目录') # 参数说明
parser.addoption('--type', # 参数名
action='store', # 表示需要存储一个值
default='', # 默认值
help='用例格式类型,yml 还是 excel') # 参数说明
# 固定名称 pytest_generate_tests,钩子函数,用于动态生成测试参数
# metafunc: pytest.Metafunc 对象,它包含了测试函数的元信息,例如参数化信息、测试函数名称等。
# metafunc.config 对象来访问 pytest 的命令行选项和配
# @pytest.mark.parametrize 和 pytest_generate_test 钩子为同一个参数提供值,那么 pytest_generate_tests 会覆盖掉
# 为不同值赋值,会同时生效,笛卡尔积
def pytest_generate_tests(self, metafunc):
case = metafunc.config.getoption('case') # 获取执行参数
type = metafunc.config.getoption('type')
case_infos, case_names = parse_pytest_parameter(case, type) # 获取指定用例目录下的用例
if 'case_info' in metafunc.fixturenames: # 测试函数所有参数名
# ids 为参数化测试用例重命名
metafunc.parametrize('case_info', case_infos, ids=case_names) # 动态生成参数
# 固定名称 pytest_generate_tests,钩子函数
# 用于在收集测试用例(还未开始执行)进行修改或操作如如更改名称,跳过某些用例,重新排序
def pytest_collection_modifyitems(self, items):
for item in items:
# 用例名称一般是测试函数名,如果提供 ids 则为 id 名称
# 解决 Unicode 转义字符的问题,正确显示中文
item.name = item.name.encode('utf-8').decode('unicode_escape')
# 节点 id,类似模块路径::测试函数名 的字符串,如 tests/test_example.py::test_example
item._nodeid = item._nodeid.encode('utf-8').decode('unicode_escape')
item.add_marker(pytest.mark.slow) # 添加标记
if 'skip' in item.name:
item.add_marker(pytest.mark.skip(reason='skip this')) # 跳过某条用例
最后再注释掉核心执行器中使用@pytest.mark.parametrize 进行用例参数化的代码
打包的目的:
框架右键执行与命令执行的区别?
工作目录可能会不同,使用的相对路径,Python 会以当前工作目录为基准来解析路径
查找模块的路径不同,导致代码中导包找不到包的路径
故命令执行与右键执行,工作目录与导包路径可能不同,导致同一份代码 可能不同时支持右键和命令运行
将框架打包,支持终端运行,打包运行命令的工作目录和导包路径都是当前运行命令的终端路径
import sys
# 方式一:sys.argv
# 是一个列表,第一个元素为脚本名称,后续元素为字符串参数
# sys.argv 不会对参数进行解析,所有参数都是字符串形式
# python main.py a=1 b=2 args=['main.py','a=1', 'b=2']
args = sys.argv
import argparse
# 方式二:使用 argparse 模块
# argparse 可以定义与解析参数
parser = argparse.ArgumentParser(description='入口执行参数')
# 1、创建解析器
# 2、添加参数,存在-(短格式),--(长格式)是可选参数,否则是必填参数
# 长短格式参数可以同时定义
parser.add_argument('-a', help='参数 a', type=int, default=0) # 可选参数
parser.add_argument('b', help='参数 b', type=str) # 必填参数,不能设置默认值
# 长短格式同时定义,执行时任选一种就行
parser.add_argument('-c', '--chlp', help='参数 b', type=str, default='name')
# 3、解析命令行参数,要解析必须定义上述添参数规则
args = parser.parse_args()
# python main1.py -a 1 hhh -c sss
print(args.a) # 1
print(args.b) # hhh
print(args.chlp) # sss,长短格式同时定义时,使用长格式参数获取
from loguru import logger
import sys
from datetime import datetime
import pytest, os, argparse
from HAT.core import core
from HAT.core.case_plugins import CasePlugin
from _pytest.config import ExitCode
import subprocess
# 配置日志
if not os.path.exists('HAT/log'):
os.mkdir('HAT/log')
time_str = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
logger.configure(handlers=[
{"sink": sys.stdout, "level": 'DEBUG'},
{"sink": sys.stderr, "level": 'DEBUG'},
{"sink": f'./log/log_{time_str}.log', "level": 'DEBUG'}
])
def args_parser():
parser = argparse.ArgumentParser(description='接口测试工具')
# 定义参数规则
parser.add_argument('--type', type=str, default='yml', help='用例文件类型')
parser.add_argument('--case', type=str, default='./test_cases/yaml_cases_01', help='指定测试用例的文件夹路径.')
parser.add_argument('--alluredir', type=str, default=os.path.join(os.getcwd(), "allure_results"), help='运行结果数据文件路径')
parser.add_argument('--allure_report', type=str, default=os.path.join(os.getcwd(), "allure_report"), help='测试报告保存路径')
args = parser.parse_args()
# 根据规则解析
return args
cmd_args = args_parser()
def run():
pytest_args = ['-vs', '--capture=sys', '--clean-alluredir']
if cmd_args.type:
pytest_args.append(f'--type={cmd_args.type}')
if cmd_args.case:
pytest_args.append(f'--case={cmd_args.case}')
if cmd_args.alluredir:
pytest_args.append(f'--alluredir={cmd_args.alluredir}')
pytest_args.append(core.__file__)
logger.info('pytest 开始执行用例')
# 返回值:0 都通过,1 存在失败,2 存在 error,5 没找到路径
exit_code = pytest.main(pytest_args, plugins=[CasePlugin()])
# OK,0 TESTS_FAILED,1
if exit_code == ExitCode.OK or exit_code == ExitCode.TESTS_FAILED:
try:
# shell 为 True,系统的 shell 执行命令,False 是接调用操作系统的 exec 系列系统调用
# check 是否以抛出异常方式运行
# universal_newlines,标准输出会被解码为字符串,否则是字节流
subprocess.run(f'allure generate {cmd_args.alluredir} -o {cmd_args.allure_report} --clean', shell=True, check=True, universal_newlines=True)
except subprocess.CalledProcessError as e:
logger.error(f"测试报告出现问题!{e}")
else:
logger.error('pytest 执行异常')
if __name__ == '__main__':
run()
setup.py 是 Python 项目打包和分发的核心配置文件,用于定义项目的元数据、依赖项和安装配置等内容,通过 setup.py,我们可以将项目打包成标准的 Python 包 方便分发和安装
setuptools 是 Python 中一个强大的库,用于打包和分发 Python 项目
在项目跟路径下增加 setup.py 文件:
import setuptools
""" 打包成一个 可执行模块 """
# with open("README.md", "r", encoding="utf-8") as fh:
# long_description = fh.read()
long_description = '自动化测试工具' # 详细描述,一般从 README.md 读取
setuptools.setup(
# 关于项目的介绍 - 随便写都可以
name="apiPlatform",
version="1.0.1",
author="author",
author_email="[email protected]",
description="自动化测试工具",
license="GPLv3",
long_description=long_description,
long_description_content_type="text/markdown",
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: GNU General Public License (GPL)",
"Operating System :: OS Independent",
],
# 需要自动安装的依赖,pin install 报名时,会自动安装这个依赖
install_requires=[
"pytest==8.4.2",
"Jinja2==3.1.6",
"jsonpath==0.82.2",
],
py_modules=["main"], # 指定单个文件,setuptools 会在跟目录下自动找到 main.py 并将其作为模块打包
packages=setuptools.find_packages(), # 扫描项目结构,自动发现项目包作为安装的一部分
python_requires=">=3.8",
entry_points={
'console_scripts': [
# tc-api-test 命令名称
# 调用 main 文件的 run 方法
'tc-api-test=main:run'
]
},
zip_safe=False # 强制以解压形式安装包
)
# 示例执行命令
# tc-api-test --type=yml --case=./test_cases/yaml_cases_01
创建文件后,执行 python setup.py install,然后就支持上述命令执行

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online
将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML转Markdown在线工具,online
通过删除不必要的空白来缩小和压缩JSON。 在线工具,JSON 压缩在线工具,online