floraachy
544 lines
20 KiB
Python
544 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
||
# @Time : 2023/9/21 17:20
|
||
# @Author : chenyinhua
|
||
# @File : data_handle.py
|
||
# @Software: PyCharm
|
||
# @Desc:
|
||
|
||
# 标准库导入
|
||
import random
|
||
import json
|
||
import re, uuid
|
||
from datetime import datetime, date, timedelta
|
||
# 第三方库导入
|
||
from loguru import logger
|
||
from faker import Faker
|
||
from string import Template
|
||
from requests.cookies import RequestsCookieJar
|
||
from requests.utils import dict_from_cookiejar
|
||
|
||
|
||
class FakerData:
|
||
"""
|
||
测试数据生成类
|
||
官方文档：https://faker.readthedocs.io/en/master/index.html
|
||
"""
|
||
|
||
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_company_name(self, lan: str = "en", fix: str = None) -> str:
|
||
"""
|
||
生成公司名
|
||
:param lan: 语言类型，可选：en, zh； zh表示中文，en表示英文，默认是en
|
||
:param fix: 前后缀，可选pre， suf； pre表示公司前缀，suf标识公司后缀
|
||
:return: 公司名
|
||
"""
|
||
if lan == "zh":
|
||
if fix == "pre":
|
||
name = self.fk_zh.company_prefix()
|
||
elif fix == "suf":
|
||
name = self.fk_zh.company_suffix()
|
||
else:
|
||
name = self.fk_zh.company()
|
||
else:
|
||
if fix == "pre":
|
||
name = self.faker.company_prefix()
|
||
elif fix == "suf":
|
||
name = self.faker.company_suffix()
|
||
else:
|
||
name = self.faker.company()
|
||
|
||
return name
|
||
|
||
def generate_paragraph(self, lan: str = "en", nb: int = 3) -> str:
|
||
"""
|
||
生成段落
|
||
:param lan: 语言类型，可选：en, zh； zh表示中文，en表示英文，默认是en
|
||
:param nb: 段落个数，默认是3个
|
||
"""
|
||
if lan == "zh":
|
||
text = self.fk_zh.paragraph(nb_sentences=nb, variable_nb_sentences=True, ext_word_list=None)
|
||
else:
|
||
text = self.faker.paragraph(nb_sentences=nb, variable_nb_sentences=True, ext_word_list=None)
|
||
|
||
return text
|
||
|
||
def generate_words(self, lan: str = "en", nb: int = 1) -> str:
|
||
|
||
"""
|
||
生成词语
|
||
:param lan: 语言类型，可选：en, zh； zh表示中文，en表示英文，默认是en
|
||
:param nb: 词语个数，默认是1个
|
||
"""
|
||
if lan == "zh":
|
||
if nb == 1 or nb < 1:
|
||
text = self.fk_zh.word(ext_word_list=None)
|
||
else:
|
||
res = self.fk_zh.words(nb=nb, ext_word_list=None)
|
||
text = "-".join(res)
|
||
|
||
else:
|
||
if nb == 1 or nb < 1:
|
||
text = self.faker.word(ext_word_list=None)
|
||
else:
|
||
res = self.faker.words(nb=nb, ext_word_list=None)
|
||
text = "-".join(res)
|
||
|
||
return text
|
||
|
||
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 process_cookie_jar(self, data):
|
||
"""
|
||
将任意数据里的RequestsCookieJar，转成dict，再转换成JSON 格式的字符串（序列化）
|
||
:param data: 待处理的数据
|
||
"""
|
||
if isinstance(data, dict):
|
||
for key, value in data.items():
|
||
data[key] = self.process_cookie_jar(value)
|
||
elif isinstance(data, list):
|
||
for i, item in enumerate(data):
|
||
data[i] = self.process_cookie_jar(item)
|
||
elif isinstance(data, RequestsCookieJar):
|
||
data = json.dumps(dict_from_cookiejar(data))
|
||
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):
|
||
obj = self.eval_data(obj)
|
||
return self.data_handle_(obj, source)
|
||
|
||
def data_handle_(self, obj, source=None):
|
||
"""
|
||
递归处理字典、列表中的字符串，将${}占位符替换成source中的值
|
||
"""
|
||
func = {}
|
||
keys = {}
|
||
if not source or not isinstance(source, dict):
|
||
logger.debug("source为空或者source不是字典格式，都将认为是：{}")
|
||
source = {}
|
||
|
||
# 处理一下source，检测到里面存在RequestsCookieJar，转成dict，再转换成JSON 格式的字符串（序列化）。
|
||
# 避免传递过来一个RequestsCookieJar，替换后变成了'RequestsCookieJar'，导致cookies无法使用的问题
|
||
source = self.process_cookie_jar(data=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)) # 替换并转为字典
|
||
|
||
should_eval = 0
|
||
for key, value in keys.items(): # 遍历字典替换
|
||
obj = obj.replace(key, value[0])
|
||
if not isinstance(source[value[1]], str):
|
||
should_eval = 1
|
||
if should_eval == 1:
|
||
obj = self.eval_data(obj)
|
||
|
||
if not isinstance(obj, str):
|
||
return self.data_handle(obj)
|
||
|
||
# 再找一遍剩余的${}跟第一步的结果合并，提取漏掉的诸如1+1的表达式(在此认为关键字无法替换的都是表达式，最后表达式也无法处理的情况就报错或者原样返回)
|
||
obj, func_temp = self.replace_and_store_placeholders(pattern, obj)
|
||
func.update(func_temp)
|
||
# 进行函数调用替换
|
||
obj = self.invoke_funcs(obj, func)
|
||
# 直接返回最后的结果
|
||
return obj
|
||
# 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:
|
||
logger.warning("Warn: --------函数：%s 无法调用成功, 请检查是否存在该函数-------" % func)
|
||
obj = obj.replace(key, funcs[0])
|
||
|
||
return obj
|
||
|
||
|
||
# 声明data_handle方法，这样外部就可以直接import data_handle来使用了
|
||
data_handle = DataHandle().data_handle
|
||
|
||
if __name__ == '__main__':
|
||
# 下面是测试代码
|
||
|
||
print("-----------测试场景1---------------------")
|
||
# 需要识别${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))
|
||
|
||
print("-----------测试场景2---------------------")
|
||
target = {
|
||
"user_id": 1,
|
||
"user_name": "flora",
|
||
"name": "test",
|
||
"age": 17
|
||
}
|
||
|
||
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))
|
||
|
||
print("-----------测试场景3---------------------")
|
||
data_03 = "user_id: ${user_id}, user_name: ${user_name}"
|
||
new = data_handle(data_03, target)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景4---------------------")
|
||
# 需要识别 字符串里面是python表达式的情况
|
||
data_04 = "[1,2,3,4]"
|
||
new = data_handle(data_04)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景5---------------------")
|
||
data_05 = "1+1"
|
||
new = data_handle(data_05)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景6---------------------")
|
||
data_06 = "[1, '1', [1, 2], {'name':'flora', 'age': '1'}]"
|
||
new = data_handle(data_06)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景7---------------------")
|
||
# 需要识别自定义的函数，同时支持多种，下面两种写法有细微差别
|
||
data_07 = "Hello, ${generate_female_name()}! Random number: ${generate_random_int()}"
|
||
new = data_handle(data_07)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景8---------------------")
|
||
data_08 = "Hello, ${generate_female_name()}! Random number: ${FakerData().generate_random_int()}"
|
||
new = data_handle(data_08)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景9---------------------")
|
||
data_09 = "Hello, ${FakerData().generate_female_name()}! Random number: ${FakerData.generate_random_int()}"
|
||
new = data_handle(data_09)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景10---------------------")
|
||
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))
|
||
|
||
print("-----------测试场景11---------------------")
|
||
data_11 = "/api/accounts/${FakerData.generate_time('%Y-%m-%d')}/login.json"
|
||
new = data_handle(data_11)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景12---------------------")
|
||
# FakerData类中没有封装random_name这个方法，会无法处理
|
||
data_12 = '[[1,2,3,4],"${FakerData().random_name()}"]'
|
||
new = data_handle(data_12)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景13---------------------")
|
||
# 下面这种写法不是字符串里面，不是正确格式的列表，${FakerData().generate_name()需要用引号包起来，因此无法正确处理成列表，最后返回的是str
|
||
data_13 = '[[1,2,3,4],${FakerData().generate_name()}]'
|
||
new = data_handle(data_13)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景14---------------------")
|
||
data_14 = '[[1,2,3,4],${FakerData().generate_random_int()}]'
|
||
new = data_handle(data_14)
|
||
print(new, type(new))
|
||
|
||
print("-----------测试场景15---------------------")
|
||
# 导入其他方法，也可以直接使用
|
||
# from common_utils.time_handle import test_fun_a
|
||
# data = "${test_fun_a()}"
|
||
# new = data_handle(data)
|
||
# print(new, type(new))
|
||
|
||
print("-----------测试场景16---------------------")
|
||
# 支付方法传参使用
|
||
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))
|
||
|
||
print("-----------测试场景17---------------------")
|
||
data_17 = {
|
||
"winner_id": "${winner_id}",
|
||
"user_id": "${user_id}",
|
||
"assignee_id": "${assignee_id}"
|
||
}
|
||
source = {
|
||
"winner_id": "1,2,4",
|
||
"assignee_id": [],
|
||
"user_id": 1076
|
||
}
|
||
|
||
res = data_handle(obj=data_17, source=source)
|
||
print(res)
|
||
|
||
print("-----------测试场景18---------------------")
|
||
payload = {'id': 'case_login_01', 'title': '用户名密码正确，登录成功(不校验数据库)', 'severity': 'NORMAL',
|
||
'url': ' https://testforgeplus.trustie.net/api/accounts/login.json', 'run': True, 'method': 'POST',
|
||
'headers': {'Content-Type': 'application/json; charset=utf-8;'}, 'cookies': None, 'request_type': 'JSON',
|
||
'payload': {'login': 'autotest', 'password': '********', 'autologin': 1}, 'files': None,
|
||
'extract': {'nickname': '$.username', 'login': '$.login', 'user_id': '$.user_id'},
|
||
'assert_response': {'eq': {'http_code': 200, '$.user_id': "${user_id}"}}, 'assert_sql': None}
|
||
|
||
source = {
|
||
"winner_id": "1,2,4",
|
||
"assignee_id": [],
|
||
"user_id": 1076
|
||
}
|
||
res = data_handle(obj=payload, source=source)
|
||
print(res)
|