基于Python+Requests+Allure搭建接口自动化框架

文章目录

• python中的*args与**kwargs
• 使用requests发送请求
• • 发送get请求
  • 发送post请求
  • 使用session发送请求
• jsonpath使用
• 框架基本封装过程
• • 目录介绍
  • yaml用例
  • • 基本格式
    • 读取yaml
  • 封装请求关键字
  • 基本核心执行器
  • main执行入口
• 核心执行器优化
• • 优化报告
  • 增加全局配置
  • • yaml用例文件改造
    • 全局变量类
  • 增加前置后置脚本执行
  • 多用例执行
  • • 用例之间进行数据关联
• 断言操作
• • 单个字段进行断言
  • 整个响应结果进行断言
  • 数据库断言
• yml数据驱动
• allure与loguru日志
• • 基本使用
  • 创建日志管理器关联步骤日志
• session复用机制
• 文件上传与下载
• • 文件上传
  • 文件下载
• pytest自定义插件
• 打包
• • mian文件内容替换
  • • python获取命令执行参数#
    • main文件内容
  • 导出框架依赖包
  • 增加setup文件

python中的*args与**kwargs

• *args：会将传入的额外位置参数打包成一个 元组（tuple）
• **kwargs：会将传入的额外关键字参数打包成一个 字典（dict）

# 调用时，参数只需传递一个一个的值，自动会打包为元组args
def my_tool(*args):
print(args)

# 调用时，参数需要传递多个 键=值，自动打包为字典kwargs
def my_tools(**kwargs):
print(kwargs)

if __name__ == '__main__':
my_tool('hello', 2, True) # ('hello', 2, True)
my_tools(name='zs', age=10) # {'name': 'zs', 'age': 10}

• 使用 ** 对字典进行解包，将键值对作为参数传递给函数
• 使用*对列表进行解包，将列表元素作为参数传递给函数

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

使用requests发送请求

需要先安装：pip install requests

发送get请求

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())

发送post请求

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参数区别

• json 要求传字典，json.loads将字符串转为字典， content-type为application/json
• data 要求字符串，json.dumps将字典转为字符串，content-type不是json一般为application/x-www-form-urlencoded
• data 可以传字典等格式数据，只是会自动编码为字符串，如果传的本身是字符串，则不进行编码

想在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')})

使用session发送请求

requests发送请求是独立的，但是有时需要保存会话内容

session = requests.session()
# 登录
ret = session.post(url=url, json=data)
# 登录后操作
ret = session.post(url=url, json=data)

• 如果使用cookie，session鉴权，使用requests.session请求
• 如果使用token鉴权，使用requests请求，手动使用token作为参数或者header传递

jsonpath使用

jsonpath用于在json数据中定位提取数据 使用前需要先安装：pip install jsonpath

规则：

• $ 根节点
• . 子节点
• .. 所有符合条件
• * 所有元素节点
• [index] 下标
• [,] 多个结果选择
• [start:end] 指定范围
• ?(@.property==value) 过滤
• @ 当前节点

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')]')

框架基本封装过程

目录介绍

• cases用例目录：存放yaml或者excel用例
• parse：读取yaml或excel用例
• core：核心执行器执行用例，本质是pytest执行
• utils：工具类和常用方法封装
• main文件：程序入口

yaml用例

基本格式

使用前安装：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

读取yaml

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文件，在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)，在函数中添加步骤

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)

• @allure.step(‘发送post请求’)，在函数外添加步骤

# 调用该函数时，自动创建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 报告里？

• tqdm 默认输出到 sys.stderr（标准错误，如异常堆栈、进度提示）
• pytest 会自动捕获测试过程中的 stderr/stdout 输出
• allure-pytest 插件会把 pytest 捕获的日志自动 attach 到 Allure 报告中，作为 stderr 或 stdout 附件

增加全局配置

yaml用例文件改造

一些全局配置如项目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用例中

全局变量类

• 属性_dict为类属性，每个实例共用一份
• 可以给实例方法上添加@classmethod，保证是类方法，里面修改会修改类属性，否则使用=进行self.attr = value赋值时，会优先在实例级别创建一个新的属性attr，而不会修改类属性
• 可以使用类名.类属性进行修改如GlobalVar._dict，而非self._dict=value
• 如果实例方法中没有进行=号赋值，直接使用类属性则会先查找实例属性_dict，找不到再查找类属性_dict

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:
assert False,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) # 保存到全局变量

yml数据驱动

修改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'))

allure与loguru日志

基本使用

安装：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)

• logger.add在系统日志配置默认基础上增加
• logger.configure(handlers=None)，handlers一个列表，列表中的每一项是一个字典，用于定义一个日志输出sink，重新配置整个logger

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
# extension='', # 附件扩展名，不提供则根据附件类型推测
) # 向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): # 报告右侧显示的步骤

日志器调用顺序梳理：

session复用机制

在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自定义插件

给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运行时会扫描加载的插件和测试代码，寻找这些钩子函数，匹配到就会在合适的时机调用它，这里使用到的钩子函数：

• pytest_addoption，用于向pytest的命令行添加自定义参数
• pytest_generate_tests，用于动态生成测试参数
• pytest_collection_modifyitems，用于在收集测试用例（还未开始执行）进行修改或操作如如更改名称，跳过某些用例，重新排序

import pytest

from requests_test.HAT.prase.yaml_prase 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进行用例参数化的代码

打包

打包的目的：

• 提高复用性：封装框架为模块，方便在其他项目中使用。
• 方便分发：通过打包和上传到 PyPI，方便团队和用户安装。
• 便于部署：简化框架的部署流程，支持标准化安装。
• 管理依赖：明确框架的依赖项，自动安装所需的库。
• 支持版本管理：通过版本号管理框架的更新和兼容性。
• 提升专业性：增强框架的可用性和标准化。
• 集成到 CI/CD：支持自动化构建、测试和发布。

框架右键执行与命令执行的区别？

工作目录可能会不同，使用的相对路径，Python会以当前工作目录为基准来解析路径

• 右键运行：默认的工作目录通常是项目的根目录，可在pycharm的Working directory中手动设置工作目录
• 命令运行：工作目录是运行命令所在的终端目录

查找模块的路径不同，导致代码中导包找不到包的路径

• 右键运行：通常会添加项目的根目录到 sys.path，以便导入项目内的模块
• 命令运行：会将当前执行文件的目录添加到sys.path[0]

故命令执行与右键执行，工作目录与导包路径可能不同，导致同一份代码 可能不同时支持右键和命令运行

将框架打包，支持终端运行，打包运行命令的工作目录和导包路径都是当前运行命令的终端路径

mian文件内容替换

python获取命令执行参数#

• sys.argv方式获取

import sys
# 方式一：sys.argv
# 是一个列表，第一个元素为脚本名称，后续元素为字符串参数
# sys.argv不会对参数进行解析，所有参数都是字符串形式
# python main.py a=1 b=2 args=['main.py','a=1', 'b=2']
args = sys.argv

• 使用argparse模块获取

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，长短格式同时定义时，使用长格式参数获取

main文件内容

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()

导出框架依赖包

• 导出，pip freeze > requirement.txt
• 安装，pip install -r requirement.txt

增加setup文件

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="hlk", # 作者
author_email="hhlongkang@163.com", # 作者邮箱
description="自动化测试工具", # 简短描述
license="GPLv3", # 许可证
long_description=long_description, # 详细描述（通常从 README.md 读取）
long_description_content_type="text/markdown", # 描述类型
url="https://gitee.com/helongkang/api_platform", # 项目主页
# 项目相关链接，如文档
project_urls={
"Bug Tracker": "https://www.hctestedu.com",
"Contact Us": "https://www.hctestedu.com",
},
# 用来分类和描述项目的性质，比如支持的编程语言、许可证类型、操作系统等
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", # 指定python安装此包的版本要求
# entry_points定义如何通过命令行运行代码
# 生成一个可执行文件 例如 windows下面 .exe
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，然后就支持上述命令执行