floraachy/apiautotest
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

进行一系列优化，且删除所有demo用例仅保持正常的示例

Browse Source
This commit is contained in:
floraachy
2023-11-10 15:58:03 +08:00
parent e97fe7290f
commit 4943adf36b
53 changed files with 906 additions and 868 deletions
Download Patch File Download Diff File Expand all files Collapse all files
case_utils/requests_utils/__init__.py
+6
View File
@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
# @Time : 2023/11/10 14:41
# @Author : floraachy
# @File : __init__.py
# @Software: PyCharm
# @Desc:
case_utils/requests_utils/base_request.py
+126
View File
@@ -0,0 +1,126 @@
# -*- coding: utf-8 -*-
# @Time : 2023/10/12 13:41
# @Author : chenyinhua
# @File : utils.py
# @Software: PyCharm
# @Desc:
# 标准库导入
import os
import time
# 第三方库导入
import allure
from loguru import logger
import requests # pip install requests
from requests_toolbelt import MultipartEncoder # pip install requests_toolbelt
class BaseRequest:
"""
Request操作封装
"""
TIMEOUT = 5
session = None
@classmethod
def get_session(cls):
"""
单例模式保证测试过程中使用的都是一个session对象；
requests.session可以自动处理cookies，做状态保持。
"""
if cls.session is None:
cls.session = requests.Session()
return cls.session
@classmethod
@allure.step("请求各项参数：{req_data}")
def send_request(cls, req_data):
"""
处理请求数据，转换成可用数据发送请求
:param req_data: 请求数据
:return: 响应对象
"""
try:
before = "\n" + "=" * 80 \
+ "\n-------------Start：请求前--------------------\n" \
f"标题: {req_data.get('title', None)}\n" \
f"请求路径: {req_data.get('url', None)}\n" \
f"请求方式: {req_data.get('method', None)}\n" \
f"请求头: {req_data.get('headers', None)}\n" \
f"请求Cookies: {req_data.get('cookies', None)}\n" \
f"请求关键字: {req_data.get('request_type', None)}\n" \
f"请求内容: {req_data.get('payload', None)}\n" \
f"请求文件: {req_data.get('file', None)}\n" \
+ "=" * 80
logger.info(before)
response_result = cls.send_api_request(
url=req_data.get("url"),
method=req_data.get("method").lower(),
request_type=req_data.get("request_type", None),
header=req_data.get("headers", None),
payload=req_data.get("payload", None),
files=req_data.get("files", None),
cookies=req_data.get("cookies", None)
)
after = "\n" + "=" * 80 \
+ "\n-------------Start：请求后--------------------\n" \
f"响应数据: {response_result.text}\n" \
f"响应码: {response_result.status_code}\n" \
f"响应耗时: {round(response_result.elapsed.total_seconds(), 2)} s || {round(response_result.elapsed.total_seconds() * 1000, 2)} ms\n" \
+ "=" * 80
logger.info(after)
return response_result
except requests.exceptions.RequestException as e:
logger.error(f"请求出错，{str(e)}")
raise ValueError(f"请求出错，{str(e)}")
@classmethod
def send_api_request(cls, url: str, method: str, request_type: str, header=None, payload=None,
files=None, cookies=None):
"""
发送请求
:param method: 请求方法
:param url: 请求url
:param request_type: 请求参数类型，可选值为params，json，data
:param payload: 请求数据，对于不同请求类型，可以为dict，MultipartEncoder等
:param files: 请求上传的文件
:param header: 请求头
:param cookies: 请求cookies
:return: 返回res对象
"""
headers = header or {}
session = cls.get_session()
if request_type:
if request_type.lower() == 'params':
res = session.request(method=method, url=url, params=payload, headers=headers, cookies=cookies,
timeout=cls.TIMEOUT)
return res
elif request_type.lower() == 'data':
res = session.request(method=method, url=url, data=payload, headers=headers, cookies=cookies,
timeout=cls.TIMEOUT)
return res
elif request_type.lower() == 'json':
res = session.request(method=method, url=url, json=payload, headers=headers, cookies=cookies,
timeout=cls.TIMEOUT)
return res
elif request_type.lower() == 'file':
if files:
file_name = os.path.basename(files)
encoder = MultipartEncoder(fields={"file": (file_name, open(files, "rb"))},
boundary='------------------------' + str(time.time()))
headers['Content-Type'] = encoder.content_type
response = session.request(method=method, url=url, data=encoder.to_string(), headers=headers,
cookies=cookies, timeout=cls.TIMEOUT)
return response
else:
logger.error("上传的文件不能为空")
else:
logger.error("request_type可选关键字为params, json, data, file")
raise ValueError('request_type可选关键字为params, json, data, file')
else:
logger.error("request_type参数不能为空")
raise ValueError('request_type参数不能为空')
case_utils/requests_utils/case_data_analysis.py
+122
View File
@@ -0,0 +1,122 @@
# -*- coding: utf-8 -*-
# @Time : 2023/6/7 16:37
# @Author : chenyinhua
# @File : case_data_analysis.py
# @Software: PyCharm
# @Desc: 分析用例数据是否符合规范
# 标准库导入
from typing import Text
# 第三方库导入
from config.models import TestCaseEnum, Method, RequestType, Severity
class CaseDataCheck:
"""
用例数据解析, 判断数据填写是否符合规范
"""
def __init__(self):
self.case_data = None
self.case_id = None
@property
def get_method(self) -> Text:
"""
处理method参数：校验method参数是否是指定枚举值中的一个
"""
return self.check_params_right(
Method,
self.case_data.get(TestCaseEnum.METHOD.value[0])
)
@property
def get_severity(self) -> Text:
"""
处理severity参数：如果Severity为空或者不传或者传错，视为NORMAL
"""
# 如果Severity为空或者不传或者传错，视为NORMAL
attr = self.case_data.get(TestCaseEnum.SEVERITY.value[0])
if attr is None or attr.upper() not in Severity._member_names_:
return "NORMAL"
else:
return attr.upper()
@property
def get_request_type(self):
"""
处理request_type参数：校验request_type参数是否是指定枚举值中的一个
"""
return self.check_params_right(
RequestType,
self.case_data.get(TestCaseEnum.REQUEST_TYPE.value[0])
)
def check_case_data_attr(self, attr: Text):
"""
检查用例中是否缺失某个参数
"""
assert attr in self.case_data.keys(), (
f"用例ID: {self.case_id} --> 缺少 {attr} 参数，请确认用例内容是否编写规范."
)
def check_params_exit(self):
"""
遍历一个枚举类中所有成员，并检查与每个成员对应的实例属性是否存在。
如果属性存在，则什么也不做，如果不存在，则抛出异常或执行其他操作
"""
for enum in list(TestCaseEnum._value2member_map_):
if enum[1]:
self.check_case_data_attr(enum[0])
def check_required_fields(self):
for field in TestCaseEnum:
if field.value[1]: # 判断是否为必填参数
field_name = field.value[0]
value = self.case_data[field_name]
if value is None or (isinstance(value, str) and len(value.strip()) == 0):
raise ValueError(f"用例ID: {self.case_id} --> {field_name}字段的值是：{value}， 该字段值是必填，请检查是否填写正确.")
def check_params_right(self, enum_name, attr):
"""
检查参数值是否正确，符合要求规范
"""
_member_names_ = enum_name._member_names_
assert attr.upper() in _member_names_, (
f"用例ID: {self.case_id} --> {enum_name}: {attr} 填写不正确，"
f"当前框架中只支持 {_member_names_} 类型."
f"如有疑问，请联系管理员."
)
return attr.upper()
def case_process(self, cases):
case_list = []
for key, values in cases.items():
# 公共配置中的数据，与用例数据不同，需要单独处理
if key == 'case_info':
for value in values:
self.case_data = value
self.case_id = value.get("id")
# 检查用例参数，需要必填的是否有值
self.check_required_fields()
# 检查用例参数值，是否都填写正确
self.check_params_exit()
case_data = {
'id': self.case_data.get(TestCaseEnum.ID.value[0]),
'title': self.case_data.get(TestCaseEnum.TITLE.value[0]),
'severity': self.get_severity,
'url': self.case_data.get(TestCaseEnum.URL.value[0]),
'run': self.case_data.get(TestCaseEnum.RUN.value[0]),
'method': self.get_method,
'headers': self.case_data.get(TestCaseEnum.HEADERS.value[0]),
'cookies': self.case_data.get(TestCaseEnum.COOKIES.value[0]),
'request_type': self.get_request_type,
'payload': self.case_data.get(TestCaseEnum.PAYLOAD.value[0]),
'files': self.case_data.get(TestCaseEnum.FILES.value[0]),
'extract': self.case_data.get(TestCaseEnum.EXTRACT.value[0]),
"assert_response": self.case_data.get(TestCaseEnum.ASSERT_RESPONSE.value[0]),
"assert_sql": self.case_data.get(TestCaseEnum.ASSERT_SQL.value[0]),
}
case_list.append(case_data)
return case_list
case_utils/requests_utils/case_fun_handle.py
+260
View File
@@ -0,0 +1,260 @@
# -*- coding: utf-8 -*-
# @Time : 2023/6/7 10:07
# @Author : chenyinhua
# @File : case_fun_handle.py
# @Software: PyCharm
# @Desc:
# 标准库导入
import os
from string import Template
import datetime
import re
# 第三方库导入
from xpinyin import Pinyin # 纯 Python 编写的中文字符串拼音转换模块，不需要依赖外部程序和词库。
from loguru import logger
# 本地应用/模块导入
from common_utils.read_files_utils.excel_handle import ExcelHandle
from common_utils.read_files_utils.yaml_handle import YamlHandle
from common_utils.other_utils.files_handle import get_files, get_relative_path
from config.models import CaseFileType
from config.settings import CASE_FILE_TYPE
from config.path_config import DATA_DIR, CASE_TEMPLATE_DIR, AUTO_CASE_DIR
from config.global_vars import CUSTOM_MARKERS
from case_utils.requests_utils.case_data_analysis import CaseDataCheck
"""
主要步骤：
1. 从用例数据文件（EXCEL/YAML）中获取用例数据
2. 分析用例数据是否符合规范
3. 确认符合规范后，获取所有用例数据，自动生成测试用例方法（PY）
"""
def handle_excel_data(file):
"""
读取excel用例数据生成测试用例
"""
if os.path.isfile(file):
# 读取excel文件中的用例数据，存储到data中
data = ExcelHandle(file).read()
for index, v in enumerate(data):
excel_case = {}
"""
将excel读取到的用例数据，适配allure格式
表单名称以短横线隔开的情况下，左边部分作为allure_epic， 右边部分作为allure_feature以及allure_story。
否则，表单名称全部作为allure_epic，allure_feature，allure_story
"""
if "-" in v["sheet_name"]:
excel_case["case_common"] = {
"allure_epic": v["sheet_name"].split("-")[0],
"allure_feature": v["sheet_name"].split("-")[1],
"allure_story": v["sheet_name"].split("-")[1]
}
else:
excel_case["case_common"] = {
"allure_epic": v["sheet_name"],
"allure_feature": v["sheet_name"],
"allure_story": v["sheet_name"]
}
# 处理excel用例数据
excel_case["case_info"] = v["data"]
# 检查用例数据是否符合规范
tested_case = CaseDataCheck().case_process(excel_case)
# 生成测试方法
"""
由于excel涉及到多个表单，每一个表单都会生成一个测试方法。因此会将表单名称的首字母拼接到测试方法上。
excel名称：test_demo
例如表单名称："GitLink-登录模块" 或 "登录模块"，都是取关键字"登录模块"首字母
测试文件：test_demo_dl.py
测试方法名称: TestDemoDl.test_demo_dl
"""
pin_yin = Pinyin()
_name = pin_yin.get_initials(excel_case["case_common"]["allure_feature"], "").lower()
if os.path.samefile(DATA_DIR, os.path.dirname(file)):
# 用例文件的直接父级目录是DATA_DIR，则直接在AUTO_CASE_DIR下生成测试用例方法
gen_case_file(
filename=os.path.splitext(os.path.basename(file))[0] + "_" + _name,
case_template_path=CASE_TEMPLATE_DIR,
case_common=excel_case["case_common"],
case_data=tested_case,
target_case_path=AUTO_CASE_DIR
)
else:
# 用例文件的直接父级目录不是DATA_DIR， 则保留其直接父级目录，再在AUTO_CASE_DIR下生成测试用例方法
gen_case_file(
filename=os.path.splitext(os.path.basename(file))[0] + "_" + _name,
case_template_path=CASE_TEMPLATE_DIR,
case_common=excel_case["case_common"],
case_data=tested_case,
target_case_path=os.path.join(AUTO_CASE_DIR,
get_relative_path(file_path=file, directory_path=DATA_DIR))
)
return True
else:
logger.error(f"{file}不是一个正确的文件路径！")
return False
def handle_yaml_data(file):
"""
读取yaml用例数据生成测试用例
"""
if os.path.isfile(file):
# 读取yaml/yml文件中的用例数据，存储到data中
yaml_data = YamlHandle(file).read_yaml
# 检查用例数据是否符合规范
tested_case = CaseDataCheck().case_process(yaml_data)
if os.path.samefile(DATA_DIR, os.path.dirname(file)):
gen_case_file(
# 用例文件的直接父级目录是DATA_DIR，则直接在AUTO_CASE_DIR下生成测试用例方法
filename=os.path.splitext(os.path.basename(file))[0],
case_template_path=CASE_TEMPLATE_DIR,
case_common=yaml_data["case_common"],
case_data=tested_case,
target_case_path=AUTO_CASE_DIR
)
else:
# 用例文件的直接父级目录不是DATA_DIR， 则保留其直接父级目录，再在AUTO_CASE_DIR下生成测试用例方法
os.makedirs(os.path.dirname(file), exist_ok=True)
gen_case_file(
filename=os.path.splitext(os.path.basename(file))[0],
case_template_path=CASE_TEMPLATE_DIR,
case_common=yaml_data["case_common"],
case_data=tested_case,
target_case_path=os.path.join(AUTO_CASE_DIR,
get_relative_path(file_path=file, directory_path=DATA_DIR))
)
return True
else:
logger.error(f"{file}不是一个正确的文件路径！")
return False
def generate_cases():
"""
根据配置文件，从指定类型文件中读取所有用例数据，并自动生成测试用例
"""
excel_files = []
yaml_files = []
if CASE_FILE_TYPE == CaseFileType.EXCEL.value:
# 在用例数据"DATA_DIR"目录中寻找后缀是xlsx, xls的文件
excel_files = get_files(target=DATA_DIR, start="test_", end=".xlsx") \
+ get_files(target=DATA_DIR, start="test_", end=".xls")
elif CASE_FILE_TYPE == CaseFileType.YAML.value:
# 在用例数据"DATA_DIR"目录中寻找后缀是yaml, yml的文件
yaml_files = get_files(target=DATA_DIR, start="test_", end=".yaml") \
+ get_files(target=DATA_DIR, start="test_", end=".yml")
elif CASE_FILE_TYPE == CaseFileType.ALL.value:
# 在用例数据"DATA_DIR"目录中寻找后缀是xlsx,xls, yaml, yml的文件
excel_files = get_files(target=DATA_DIR, start="test_", end=".xlsx") + get_files(target=DATA_DIR, start="test_",
end=".xls")
yaml_files = get_files(target=DATA_DIR, start="test_", end=".yaml") + get_files(target=DATA_DIR, start="test_",
end=".yml")
else:
logger.error(f"{CASE_FILE_TYPE}不在CaseFileType内，不能自动生成用例！")
# 自动生成测试用例
for file in excel_files:
handle_excel_data(file=file)
for file in yaml_files:
handle_yaml_data(file=file)
def gen_case_file(filename, case_template_path, case_common, case_data, target_case_path):
"""
根据测试用例文件(yaml/yml/xlsx/xls)，以及事先定义的测试用例模板，实际用例数据，生成测试用例方法(.py)
:param filename: 测试用例文件(yaml/yml/xlsx/xls)的名称，用作生成测试用例类名，方法名
:param case_template_path: 测试用例模板的绝对路径
:param case_common: 用例公共参数
:param case_data: 实际用例数据
:param target_case_path: 测试用例方法(.py)的绝对路径
"""
# 如果自动生成用例的目录不存在则自动创建一个
if not os.path.exists(target_case_path):
"""
exist_ok=True 是一个可选参数，用于指定在目录已经存在的情况下是否忽略错误。
如果设置为 True，则不论目录是否已存在，os.makedirs 都不会报错；如果设置为 False（默认值），则在目录已存在时会引发 FileExistsError 异常。
"""
os.makedirs(target_case_path, exist_ok=True)
# 获取用例数据中的标记
case_markers = case_common.get("case_markers", []) or []
logger.trace(f"从用例中拿到的标记有：{case_markers}， {type(case_markers)}")
# 先读取用例模板中每一行的内容
with open(file=case_template_path, mode="r", encoding="utf-8") as f:
case_template = f.readlines()
current_case_template = []
for line_num, content in enumerate(case_template):
current_case_template.append(content)
# 这里是预计往 @pytest.mark.parametrize( 这一行的上面插入标记
if content.strip().startswith('@pytest.mark.parametrize('):
# 往测试用例模板中插入自定义标记
logger.trace(f"获取到的case_markers：{case_markers}， {type(case_markers)}")
for case_marker in case_markers:
# 获取符合要求格式的自定义标记名称，并插入到测试模板中
marker = is_valid_marker(case_marker)
if marker and isinstance(marker, str):
# ！！ 注意这里的4个空格，必须要有4个空格！！
current_case_template.append(f" @pytest.mark.{marker}\n")
if marker and isinstance(marker, dict):
for k, v in marker.items():
# ！！ 注意这里的4个空格，必须要有4个空格！！
current_case_template.append(f" @pytest.mark.{k}('{v}')\n")
# 将用例数据的名称作为测试用例文件名称, 如test_login_demo
func_name = filename
# 方法名test_demo的类名是TestDemo
class_name = "".join([word.capitalize() for word in func_name.split("_")])
# 根据模板，生成测试用例方法
"""
string.Template是将一个string设置为模板，通过替换变量的方法，最终得到想要的string。
"""
current_template = ''.join(current_case_template)
my_case = Template(current_template).safe_substitute(
{
"allure_epic": case_common["allure_epic"],
"allure_feature": case_common["allure_feature"],
"allure_story": case_common["allure_story"],
"case_markers": case_markers,
"case_data": case_data,
"func_title": func_name,
"class_title": class_name,
"now": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
)
# 将测试用例方法写入py文件中
with open(os.path.join(target_case_path, func_name + '.py'), "w", encoding="utf-8") as fp:
fp.write(my_case)
def is_valid_marker(markers):
"""
检查标记名称是否合法：仅支持非数字/下划线开头，由数字，字母，下划线组成的标记名称
"""
pattern = r'^[a-zA-Z_][a-zA-Z0-9_]*$'
if isinstance(markers, str):
if re.match(pattern, markers):
# 将自定义标记放到CUSTOM_MARKERS， 方便后续统一注册
if markers not in ("skip", "skipif", "parametrize", "usefixtures", "xfail", "filterwarings"):
CUSTOM_MARKERS.append(markers)
# 返回合法有效的标记名称，用于添加到测试方法中
return markers
else:
logger.error(f"{markers} 格式不合法， 建议仅输入数字，字母，下划线组合，且不能以数字，下划线开头")
return False
elif isinstance(markers, dict):
if len(markers) == 1:
marker_name = list(markers.keys())[0]
if re.match(pattern, marker_name):
# 将自定义标记放到CUSTOM_MARKERS， 方便后续统一注册
if marker_name not in ("skip", "skipif", "parametrize", "usefixtures", "xfail", "filterwarings"):
CUSTOM_MARKERS.append(markers)
return markers
else:
logger.error(f"{markers} 格式不合法， 建议仅输入数字，字母，下划线组合，且不能以数字，下划线开头")
return None
else:
logger.error(f"{markers} 格式不合法， 只能存在一对键值对")
return None
else:
logger.error(f"{markers} 仅支持字符串或者字典格式")
return None
case_utils/requests_utils/data_handle.py
+397
View File
@@ -0,0 +1,397 @@
# -*- coding: utf-8 -*-
# @Time : 2023/9/21 17:20
# @Author : chenyinhua
# @File : data_handle.py
# @Software: PyCharm
# @Desc:
# 标准库导入
import random
import re, uuid
from datetime import datetime, date, timedelta
# 第三方库导入
from faker import Faker
from string import Template
class FakerData:
"""
测试数据生成类
"""
def __init__(self):
self.fk_zh = Faker(locale='zh_CN')
self.faker = Faker()
@classmethod
def generate_random_int(cls, *args) -> int:
"""
:return: 随机数
"""
# 检查是否传入了参数
if not args:
# 没有传参，就从5000内随机取一个整数返回
return random.randint(0, 5000)
# 排序参数并获取最小值和最大值
min_val = min(args)
max_val = max(args)
# 生成并返回随机整数
return random.randint(min_val, max_val)
def generate_phone(self, lan="en") -> int:
"""
:return: 随机生成手机号码
"""
if lan == "zh":
phone = self.fk_zh.phone_number()
else:
phone = self.faker.phone_number()
return phone
def generate_id_number(self, lan="en") -> int:
"""
:return: 随机生成身份证号码
"""
if lan == "zh":
id_number = self.fk_zh.ssn()
else:
id_number = self.faker.ssn()
return id_number
def generate_female_name(self, lan="en") -> str:
"""
:return: 女生姓名
"""
if lan == "zh":
female_name = self.fk_zh.name_female()
else:
female_name = self.faker.name_female()
return female_name
def generate_male_name(self, lan="en") -> str:
"""
:return: 男生姓名
"""
if lan == "zh":
male_name = self.fk_zh.name_male()
else:
male_name = self.faker.name_male()
return male_name
def generate_name(self, lan="en") -> str:
"""
:return: 人名
"""
if lan == "zh":
name = self.fk_zh.name()
else:
name = self.faker.name()
return name
def generate_email(self, lan="en") -> str:
"""
:return: 生成邮箱
"""
if lan == "zh":
email = self.fk_zh.email()
else:
email = self.faker.email()
return email
def generate_identifier(self, lan="en"):
"""
:return:生成随机标识，满足要求：长度为2~100， 只能包含数字，字母，下划线(_)，中划线(-)，英文句号(.)，必须以数字和字母开头，不能以下划线/中划线/英文句号开头和结尾
"""
if lan == "zh":
fk = self.fk_zh
else:
fk = self.faker
while True:
identifier = fk.slug() # 生成随机的slug标识
if (
re.match(r'^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,98}[a-zA-Z0-9]$', identifier) and
not (identifier.startswith('_') or identifier.startswith('-') or identifier.startswith('.')) and
not (identifier.endswith('_') or identifier.endswith('.'))
):
return identifier
@classmethod
def generate_time(cls, fmt='%Y-%m-%d %H:%M:%S') -> str:
"""
计算当前时间
:return:
"""
now_time = datetime.now().strftime(fmt)
return now_time
@classmethod
def generate_today_date(cls):
"""获取今日0点整时间"""
_today = date.today().strftime("%Y-%m-%d") + " 00:00:00"
return str(_today)
@classmethod
def generate_time_after_week(cls):
"""获取一周后12点整的时间"""
_time_after_week = (date.today() + timedelta(days=+6)).strftime("%Y-%m-%d") + " 00:00:00"
return _time_after_week
@classmethod
def remove_special_characters(cls, target: str):
"""
移除字符串中的特殊字符。
在Python中用replace()函数操作指定字符
常用字符unicode的编码范围：
数字：\u0030-\u0039
汉字：\u4e00-\u9fa5
大写字母：\u0041-\u005a
小写字母：\u0061-\u007a
英文字母：\u0041-\u007a
"""
pattern = r'([^\u4e00-\u9fa5])'
result = re.sub(pattern, '', target)
return result
class DataHandle:
def __init__(self):
# 实例化FakerData类，避免反复实例，提高性能。
self.FakerDataClass = FakerData()
# 获取FakerData类所有自定义方法
self.method_list = [method for method in dir(FakerData) if
callable(getattr(FakerData, method)) and not method.startswith("__")]
# 将"[1,2,3]" 或者"{'k':'v'}" -> [1,2,3], {'k':'v'}
def eval_data(self, data):
"""
执行一个字符串表达式，并返回其表达式的值
"""
try:
if hasattr(eval(data), "__call__"):
return data
else:
return eval(data)
except Exception:
return data
def replace_and_store_placeholders(self, pattern, text):
"""
提取字符串中符合正则表达式的元素，同时用一个唯一的uuid来替换原有字符串
例如：
原字符串：user_id: ${user_id}, user_name: ${user_name}
替换后的原字符串：user_id: e1c6fc74-2f21-49a9-8d0c-de16650c6364, user_name: 50c74155-5cb5-4809-bc5d-277addf8c3e7
暂存的需要被处理的关键字或函数：{'e1c6fc74-2f21-49a9-8d0c-de16650c6364': {0: '${user_id}', 1: 'user_id'}, '50c74155-5cb5-4809-bc5d-277addf8c3e7': {0: '${user_name}', 1: 'user_name'}}
"""
placeholders = {}
def replace(match):
placeholder = str(uuid.uuid4()) # 使用uuid生成唯一的占位符
placeholders[placeholder] = {0: match.group(0), 1: match.group(1)} # 将提取到的字符串存储到字典中
return placeholder
# 使用正则表达式进行字符串匹配和替换，同时指定替换次数为 1
replaced_text = re.sub(pattern, replace, text, count=1)
while replaced_text != text:
text = replaced_text
replaced_text = re.sub(pattern, replace, text, count=1)
return replaced_text, placeholders
def data_handle(self, obj, source=None):
"""
递归处理字典、列表中的字符串，将${}占位符替换成source中的值
"""
func = {}
keys = {}
if not source or not isinstance(source, dict):
# print("source为空或者source不是字典格式，都将认为是：{}")
source = {}
# 如果进来的是字符串，先将各种类型的表达式处理完
if isinstance(obj, str):
# 先把python表达式找出来存着，这里会漏掉一些诸如1+1的表达式
pattern = r"\${([^}]+\))}" # 匹配以 "${" 开头、以 ")}" 结尾的字符串，并在括号内提取内容，括号内不能包含"}"字符
obj, func = self.replace_and_store_placeholders(pattern, obj)
# 再把关键字替换的找出来存着，这里会将1+1这样的表达式存起来
pattern = r'\$\{([^}]+)\}' # 定义匹配以"${"开头，"}"结尾的字符串的正则表达式
obj, keys = self.replace_and_store_placeholders(pattern, obj)
# 接着处理表达式和关键字替换，先进行关键字替换
keys = eval(Template(str(keys)).safe_substitute(source)) # 替换并转为字典
for key, value in keys.items(): # 遍历字典替换
obj = obj.replace(key, value[0])
# 再找一遍剩余的${}跟第一步的结果合并，提取漏掉的诸如1+1的表达式(在此认为关键字无法替换的都是表达式，最后表达式也无法处理的情况就报错或者原样返回)
obj, func_temp = self.replace_and_store_placeholders(pattern, obj)
func.update(func_temp)
# 进行函数调用替换
obj = self.invoke_funcs(obj, func)
# 把处理后的结果，eval一下
return self.eval_data(obj)
elif isinstance(obj, list):
for index, item in enumerate(obj):
obj[index] = self.data_handle(item, source)
return obj
elif isinstance(obj, dict):
for key, value in obj.items():
obj[key] = self.data_handle(value, source)
return obj
else:
return obj
def invoke_funcs(self, obj, funcs):
"""
调用方法，并将方法返回的结果替换到obj中去
"""
for key, funcs in funcs.items(): # 遍历方法字典调用并替换
func = funcs[1]
# print("invoke func : ", func)
try:
if "." in func:
if func.startswith("faker."):
# 英文的faker数据：self.faker = Faker()
faker = self.FakerDataClass.faker
obj = obj.replace(key, eval(func))
elif func.startswith("fk_zh."):
# 中文的faker数据： self.fk_zh = Faker(locale='zh_CN')
fk_zh = self.FakerDataClass.fk_zh
obj = obj.replace(key, eval(func))
else:
obj = obj.replace(key, str(eval(func)))
else:
func_parts = func.split('(')
func_name = func_parts[0]
func_args_str = ''.join(func_parts[1:])[:-1]
if func_name in self.method_list: # 证明是FakerData类方法
method = getattr(self.FakerDataClass, func_name)
res = eval(f"method({func_args_str})") # 尝试直接调用
obj = obj.replace(key, str(res))
else: # 不是FakerData类方法，但有可能是 1+1 这样的
obj = obj.replace(key, str(eval(func)))
except:
print("Warn: --------函数：%s 无法调用成功, 请检查是否存在该函数-------" % func)
obj = obj.replace(key, funcs[0])
pass
return obj
# 声明data_handle方法，这样外部就可以直接import data_handle来使用了
data_handle = DataHandle().data_handle
if __name__ == '__main__':
# 下面是测试代码
target = {
"user_id": 1,
"user_name": "flora",
"name": "test",
"age": 17
}
# 需要识别${python表达式}，这里random方法是需要导入random包的
data_01 = "选择.gitignore: ${random.choice(['Ada', 'Actionscript', 'Ansible', 'Android', 'Agda'])}，开源许可证: ${random.choice(['0BSD', 'AAL', 'AFL-1.1', '389-exception'])}"
# new = data_handle(data_01)
# print(new, type(new))
# 这个用到了target
data_02 = {
"age": "${generate_random_int()}.",
"message": "Hello, ${FakerData().generate_female_name()}! Your age is ${age}. Random number: ${FakerData().generate_random_int()}",
"nested_data": [
"This is ${name}'s data.",
{
"message": "Age: ${age}.",
"nested_list": [
"More data: ${generate_random_int()}",
]
}
]
}
# new = data_handle(data_02, target)
# print(new, type(new))
data_03 = "user_id: ${user_id}, user_name: ${user_name}"
# new = data_handle(data_03, target)
# print(new, type(new))
# 需要识别 字符串里面是python表达式的情况
data_04 = "[1,2,3,4]"
# new = data_handle(data_04)
# print(new, type(new))
data_05 = "1+1"
# new = data_handle(data_05)
# print(new, type(new))
data_06 = "[1, '1', [1, 2], {'name':'flora', 'age': '1'}]"
# new = data_handle(data_06)
# print(new, type(new))
# 需要识别自定义的函数，同时支持多种，下面两种写法有细微差别
data_07 = "Hello, ${generate_female_name()}! Random number: ${generate_random_int()}"
# new = data_handle(data_07)
# print(new, type(new))
data_08 = "Hello, ${generate_female_name()}! Random number: ${FakerData().generate_random_int()}"
# new = data_handle(data_08)
# print(new, type(new))
data_09 = "Hello, ${FakerData().generate_female_name()}! Random number: ${FakerData.generate_random_int()}"
# new = data_handle(data_09)
# print(new, type(new))
data_10 = {
"payload": {
"startTime": "${FakerData.generate_time('%Y-%m-%d')}",
"common2": "${faker.name()}", # 这里是使用类FakerData里面的实例属性faker
"url": "/api/accounts/${FakerData.generate_time('%Y-%m-%d')} / login.json",
"fragement": {
"startTime": "${FakerData.generate_time('%Y-%m-%d')}",
"common2": "${faker.name()}", # 这里是使用类FakerData里面的实例属性faker
"url": "/api/accounts/${FakerData.generate_time('%Y-%m-%d')} / login.json"
}
}
}
# new = data_handle(data_10)
# print(new, type(new))
data_11 = "/api/accounts/${FakerData.generate_time('%Y-%m-%d')}/login.json"
# new = data_handle(data_11)
# print(new, type(new))
# FakerData类中没有封装random_name这个方法，会无法处理
data_12 = '[[1,2,3,4],"${FakerData().random_name()}"]'
# new = data_handle(data_12)
# print(new, type(new))
# 下面这种写法不是字符串里面，不是正确格式的列表，${FakerData().generate_name()需要用引号包起来，因此无法正确处理成列表，最后返回的是str
data_13 = '[[1,2,3,4],${FakerData().generate_name()}]'
# new = data_handle(data_13)
# print(new, type(new))
data_15 = '[[1,2,3,4],${FakerData().generate_random_int()}]'
# new = data_handle(data_15)
# print(new, type(new))
# 导入其他方法，也可以直接使用
# from common_utils.time_handle import test_fun_a
# data = "${test_fun_a()}"
# new = data_handle(data)
# print(new, type(new))
# 支付方法传参使用
payload = {
"name": "${generate_name(lan='zh')}",
"repository_name": "${generate_name('zh')}",
"desc": '[[1,2,3,4],"${FakerData().generate_random_int()}"]',
"pre": '[[1,2,3,4],${FakerData().generate_name()}]'
}
# new = data_handle(payload)
# print(new, type(new))
case_utils/requests_utils/extract_data_handle.py
+60
View File
@@ -0,0 +1,60 @@
# -*- coding: utf-8 -*-
# @Time : 2023/6/28 14:35
# @Author : chenyinhua
# @File : extract_data_handle.py
# @Software: PyCharm
# @Desc: 提取数据的一些方法
# 标准库导入
import re
# 第三方库导入
from jsonpath import jsonpath
from loguru import logger
def json_extractor(obj: dict, expr: str = '.'):
"""
:param obj :json/dict类型数据
:param expr: 表达式, . 提取字典所有内容， $.test_api_case 提取一级字典case， $.test_api_case.data 提取case字典下的data
:return result: 提取的结果，未提取到返回 None
"""
try:
result = jsonpath(obj, expr)[0]
logger.trace("\n======================================================\n" \
"-------------Start：json_extractor--------------------\n"
f"提取表达式为： {expr} \n"
f"提取值为： {result}\n"
"=====================================================")
except Exception as e:
logger.trace("\n======================================================\n" \
"-------------End：json_extractor--------------------\n"
f"提取表达式为： {expr}\n"
f"提取数据为： {obj}\n"
f"错误信息为：{e}\n"
"=====================================================")
result = None
return result
def re_extract(obj: str, expr: str = '.'):
"""
:param obj : 字符串数据
:param expr: 正则表达式
:return result: 提取的结果，未提取到返回 None
"""
try:
result = re.findall(expr, obj)[0]
logger.trace("\n======================================================\n" \
"-------------Start：re_extract--------------------\n"
f"提取表达式为： {expr}\n" \
f"提取值为： {result}\n" \
"=====================================================")
except Exception as e:
logger.trace(f"\n======================================================\n" \
"-------------End：re_extract--------------------\n"
f"提取表达式为： {expr}\n" \
f"提取数据为： {obj}\n" \
f"错误信息为：{e}\n" \
"=====================================================")
result = None
return result
case_utils/requests_utils/request_data_handle.py
+261
View File
@@ -0,0 +1,261 @@
# -*- coding: utf-8 -*-
# @Version: Python 3.9
# @Time : 2023/1/31 14:31
# @Author : chenyinhua
# @File : request_data_handle.py
# @Software: PyCharm
# @Desc: 处理request请求前后的用例数据
# 标准库导入
import json
import os
import http.cookiejar
# 第三方库导入
from requests import Response
from loguru import logger
# 本地应用/模块导入
from common_utils.other_utils.files_handle import get_file_field
from case_utils.requests_utils.base_request import BaseRequest
from case_utils.requests_utils.data_handle import data_handle
from case_utils.requests_utils.extract_data_handle import json_extractor, re_extract
from case_utils.report_utils.allure_handle import allure_step
from config.global_vars import GLOBAL_VARS
from config.path_config import FILES_DIR
# ---------------------------------------- 请求前的数据处理----------------------------------------#
class RequestPreDataHandle:
"""
请求前处理用例数据
"""
def __init__(self, request_data):
logger.trace(f"\n======================================================\n" \
"-------------Start：处理用例数据前--------------------\n"
f"用例标题(title): {type(request_data.get('title', None))} || {request_data.get('title', None)}\n" \
f"用例优先级(severity): {type(request_data.get('severity', None))} || {request_data.get('severity', None)}\n" \
f"请求域名(host): {type(request_data.get('host', None))} || {request_data.get('host', None)}\n" \
f"请求路径(url): {type(request_data.get('url', None))} || {request_data.get('url', None)}\n" \
f"请求方式(method): {type(request_data.get('method', None))} || {request_data.get('method', None)}\n" \
f"请求头(headers): {type(request_data.get('headers', None))} || {request_data.get('headers', None)}\n" \
f"请求cookies: {type(request_data.get('cookies', None))} || {request_data.get('cookies', None)}\n" \
f"请求类型(request_type): {type(request_data.get('request_type', None))} || {request_data.get('request_type', None)}\n" \
f"请求参数(payload): {type(request_data.get('payload', None))} || {request_data.get('payload', None)}\n" \
f"请求文件(files): {type(request_data.get('files', None))} || {request_data.get('files', None)}\n" \
f"后置提取参数(extract): {type(request_data.get('extract', None))} || {request_data.get('extract', None)}\n" \
f"响应断言(assert_response): {type(request_data.get('assert_response', None))} || {request_data.get('assert_response', None)}\n" \
f"数据库断言(assert_sql): {type(request_data.get('assert_sql', None))} || {request_data.get('assert_sql', None)}\n" \
"=====================================================")
self.request_data = request_data
def request_data_handle(self):
"""
针对用例数据进行处理，识别用例数据中的关键字${xxxx}，使用全局变量进行替换或者执行关键字中的方法替换为具体值
"""
self.url_handle()
self.method_handle()
self.headers_handle()
self.cookies_handle()
self.payload_handle()
self.files_handle()
self.assert_handle()
logger.trace(f"\n======================================================\n" \
"-------------End：处理用例数据后--------------------\n"
f"用例标题(title): {type(self.request_data.get('title', None))} || {self.request_data.get('title', None)}\n" \
f"用例优先级(severity): {type(self.request_data.get('severity', None))} || {self.request_data.get('severity', None)}\n" \
f"请求路径(url): {type(self.request_data.get('url', None))} || {self.request_data.get('url', None)}\n" \
f"请求方式(method): {type(self.request_data.get('method', None))} || {self.request_data.get('method', None)}\n" \
f"请求头(headers): {type(self.request_data.get('headers', None))} || {self.request_data.get('headers', None)}\n" \
f"请求cookies: {type(self.request_data.get('cookies', None))} || {self.request_data.get('cookies', None)}\n" \
f"请求类型(request_type): {type(self.request_data.get('request_type', None))} || {self.request_data.get('request_type', None)}\n" \
f"请求参数(payload): {type(self.request_data.get('payload', None))} || {self.request_data.get('payload', None)}\n" \
f"请求文件(files): {type(self.request_data.get('files', None))} || {self.request_data.get('files', None)}\n" \
f"后置提取参数(extract): {type(self.request_data.get('extract', None))} || {self.request_data.get('extract', None)}\n" \
f"响应断言(assert_response): {type(self.request_data.get('assert_response', None))} || {self.request_data.get('assert_response', None)}\n" \
f"数据库断言(assert_sql): {type(self.request_data.get('assert_sql', None))} || {self.request_data.get('assert_sql', None)}\n" \
"=====================================================")
return self.request_data
def url_handle(self):
"""
用例数据中获取到的url(一般是不带host的，个别特殊的带有host，则不进行处理)
"""
# 检测url中是否存在需要替换的参数，如果存在则进行替换
url = data_handle(obj=self.request_data.get("url", None), source=GLOBAL_VARS)
# 进行url处理，最终得到full_url
host = GLOBAL_VARS.get("host", "")
# 从用例数据中获取url，如果键url不存在，则返回空字符串
# 如果url是以http开头的，则直接使用该url，不与host进行拼接
if url.lower().startswith("http"):
full_url = url
else:
# 如果host以/结尾 并且 url以/开头
if host.endswith("/") and url.startswith("/"):
full_url = host[0:len(host) - 1] + url
# 如果host以/结尾 并且 url不以/开头
elif host.endswith("/") and (not url.startswith("/")):
full_url = host + url
elif (not host.endswith("/")) and url.startswith("/"):
# 如果host不以/结尾 且 url以/开头，则将host和url拼接起来，组成新的url
full_url = host + url
else:
# 如果host不以/结尾 且 url不以/开头，则将host和url拼接起来的时候增加/，组成新的url
full_url = host + "/" + url
self.request_data["url"] = full_url
def method_handle(self):
# TODO 暂时不需要处理，后续有需要在处理
pass
def cookies_handle(self):
"""
requests模块中，cookies参数要求是Dict or CookieJar object
"""
cookies = self.request_data.get("cookies", None)
# 从用例数据中获取cookies， 处理cookies
if cookies:
logger.debug(f"打印一下全局变量的值：{GLOBAL_VARS}")
# 通过全局变量替换cookies，得到的是一个str类型
cookies = data_handle(obj=cookies, source=GLOBAL_VARS)
try:
cookies = json.loads(cookies)
except Exception as e:
cookies = cookies
logger.debug(f"处理{cookies}报错了：{e}")
if isinstance(cookies, dict) or isinstance(cookies, http.cookiejar.CookieJar):
self.request_data["cookies"] = cookies
else:
logger.error(
f"cookies参数要求是Dict or CookieJar object， 目前cookies类型是：{type(cookies)}， cookies值是：{cookies}")
raise TypeError(
f"cookies参数要求是Dict or CookieJar object， 目前cookies类型是：{type(cookies)}， cookies值是：{cookies}")
def headers_handle(self):
"""
headers里面传cookies，要求cookies类型是str
"""
headers = self.request_data.get("headers", None)
# 从用例数据中获取header， 处理header
if headers:
self.request_data["headers"] = data_handle(obj=headers, source=GLOBAL_VARS)
# 如果请求头中有cookies，需要进行单独处理
if self.request_data["headers"].get("cookies", None):
cookies = self.request_data["headers"]["cookies"]
if isinstance(cookies, dict):
# 如果是字典类型，就转成字符串
self.request_data["headers"]["cookies"] = json.dumps(cookies)
else:
self.request_data["headers"]["cookies"] = cookies
def payload_handle(self):
# 处理请求参数payload
payload = self.request_data.get("payload", None)
if payload:
self.request_data["payload"] = data_handle(obj=payload, source=GLOBAL_VARS)
def files_handle(self):
"""
格式：接口中文件参数的名称:"文件路径地址"/["文件地址1", "文件地址2"]
例如：{"file": "demo_test_demo.py"}
或者
{"file": ["test_demo_01.py", "test_demo_02.py"]}
"""
# 处理请求参数files参数
files = self.request_data.get("files", None)
if files:
for k, v in files.items():
# ------------------ 单文件 ------------------
# 处理文件绝对路径
file_path = os.path.join(FILES_DIR, v)
# 单文件上传需要是字典{'file': (filename, file_content)}
self.request_data["files"] = {k: get_file_field(file_path)}
logger.debug(f"处理完成后的files:{self.request_data['files']}")
def assert_handle(self):
# 处理响应断言参数
assert_response = self.request_data.get("assert_response", None)
if assert_response:
self.request_data["assert_response"] = data_handle(obj=assert_response, source=GLOBAL_VARS)
# 由于数据库断言里面的变量需要请求响应后进行提取，因此目前不进行处理
# ---------------------------------------- 进行请求，请求后的参数提取处理----------------------------------------#
class RequestHandle:
"""
进行请求，请求后的参数提取处理
"""
def __init__(self, case_data):
self.case_data = case_data
def http_request(self):
"""
发送请求并进行后置参数提取操作
"""
response = BaseRequest.send_request(self.case_data)
allure_step(f"响应数据: {response.text}")
allure_step(f"响应码: {response.status_code}")
allure_step(
f"响应耗时: {round(response.elapsed.total_seconds(), 2)} s || {round(response.elapsed.total_seconds() * 1000, 2)} ms")
# 处理数据库断言 - 从全局变量中获取最新值，替换数据库断言中的参数
if self.case_data.get('assert_sql', None):
self.case_data["assert_sql"] = data_handle(obj=self.case_data["assert_sql"], source=GLOBAL_VARS)
# 处理请求里面的files，使得日志以及allure中写入的是文件，而不是文件二进制内容
if self.case_data.get('files', None):
files = self.case_data["files"]
if isinstance(files, dict):
dict_values = list(files.values())[0]
_file = os.path.join(FILES_DIR, dict_values[0])
logger.info(
f"\n请求文件(files): {type(_file)} || {_file}\n")
return response
# ---------------------------------------- 请求后的参数提取处理----------------------------------------#
def after_request_extract(response: Response, extract):
"""
从响应数据中提取请求后的参数，并保存到全局变量中
:param response: request 响应对象
:param extract: 需要提取的参数字典 '{"k1": "$.data"}' 或 '{"k1": "data:(.*?)$"}'
:return:
"""
logger.debug(f"\n======================================================\n" \
"-------------Start：从响应数据中提取后置参数保存到全局变量--------------------\n"
f"后置提取参数（原）: {extract}\n" \
"=====================================================")
result = {}
if extract:
if response_type(response) == "json":
# 如果响应数据是json格式，则将按照json方式对后置提取参数进行处理
res = response.json()
for k, v in extract.items():
result[k] = json_extractor(res, v)
else:
# 如果响应数据是str格式，则将按照str方式对后置提取参数进行处理
res = response.text
for k, v in extract.items():
result[k] = re_extract(res, v)
logger.debug(f"\n======================================================\n" \
"-------------End：从响应数据中提取后置参数保存到全局变量--------------------\n"
f"后置提取参数（新）: {result}\n" \
"=====================================================")
# 将提取到的变量保存在全局变量中
if result:
for k, v in result.items():
GLOBAL_VARS[k] = v
return result
def response_type(response: Response) -> str:
"""
:param response: requests 返回
:return: 返回响应数据类型 json或者str
"""
try:
response.json()
return "json"
except:
return "str"