增加数据处理及通知模板匹配功能
This commit is contained in:
wangjie
parent
9ffdd83d5c
commit
b1a8609836
|
|
@ -46,3 +46,7 @@ class SendMessageError(MyBaseFailure):
|
|||
|
||||
class ValueNotFoundError(MyBaseFailure):
|
||||
pass
|
||||
|
||||
|
||||
class DataProcessorFuncError(MyBaseFailure):
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -5,13 +5,7 @@
|
|||
# @File : models.py
|
||||
# @project : SensoroApi
|
||||
|
||||
# 标准库导入
|
||||
import types
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum, unique # python 3.x版本才能使用
|
||||
from typing import Text, Dict, Union, Any, Optional, List, Callable
|
||||
# 第三方库导入
|
||||
from pydantic import BaseModel
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class Environment(Enum):
|
||||
|
|
@ -21,8 +15,6 @@ class Environment(Enum):
|
|||
DIANJUN = 'dianjun'
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print(Environment.DEV.name)
|
||||
print(Environment.DEV.value)
|
||||
|
|
|
|||
|
|
@ -7,15 +7,15 @@
|
|||
|
||||
from common.models import Environment
|
||||
from utils.jenkins_handle import ProjectName, BUILD_NUMBER, ALLURE_URL, BUILD_URL
|
||||
from utils.reportdatahandle import ReportDataHandle
|
||||
|
||||
# 设置运行的环境变量
|
||||
# ------------------------------------ 通用配置 ----------------------------------------------------#
|
||||
'''
|
||||
开发环境：Environment.DEV
|
||||
测试环境：Environment.TEST
|
||||
生产环境：Environment.PROD
|
||||
点军环境：Environment.DIANJUN
|
||||
'''
|
||||
# 设置运行环境
|
||||
ENV = Environment.TEST
|
||||
|
||||
# 设置是否需要发送邮件：Ture发送，False不发送
|
||||
|
|
@ -32,7 +32,7 @@ LOG_CONSOLE = True
|
|||
|
||||
# ------------------------------------ 邮件配置信息 ----------------------------------------------------#
|
||||
|
||||
# 发送邮件的相关配置信息
|
||||
# 发送邮件配置信息
|
||||
email_config = {
|
||||
'mail_subject': '接口自动化测试报告', # 邮件标题
|
||||
'sender_username': 'xxxxx@qq.com', # 发件人邮箱
|
||||
|
|
@ -42,53 +42,52 @@ email_config = {
|
|||
'smtp_port': 465, # 发送邮箱的端口号
|
||||
}
|
||||
|
||||
# TODO:使用模板替换的方式替换该部分内容，否则此处会读取上次执行的报告
|
||||
# 邮件通知内容
|
||||
pytest_result = ReportDataHandle.pytest_json_report_case_count()
|
||||
email_content = f"""
|
||||
email_content = """
|
||||
各位同事, 大家好:<br>
|
||||
|
||||
自动化用例于 <strong>{pytest_result["case_start_time"]}</strong> 开始运行，运行时长：<strong>{pytest_result['case_duration']}s</strong>， 目前已执行完成。<br>
|
||||
自动化用例于 <strong>${case_start_time}</strong> 开始运行，运行时长：<strong>${case_duration}s</strong>， 目前已执行完成。<br>
|
||||
---------------------------------------------------------------------------------------------------------------<br>
|
||||
项目名称：<strong>{ProjectName}</strong> <br>
|
||||
构件编号：<strong>#{BUILD_NUMBER}</strong><br>
|
||||
项目环境：<strong>{ENV.name}</strong><br>
|
||||
项目名称：<strong>%s</strong> <br>
|
||||
构件编号：<strong>#%s</strong><br>
|
||||
项目环境：<strong>%s</strong><br>
|
||||
---------------------------------------------------------------------------------------------------------------<br>
|
||||
执行结果如下:<br>
|
||||
用例运行总数:<strong> {pytest_result['total_case']}条</strong><br>
|
||||
通过用例个数（passed）: <strong><font color="green" >{pytest_result['pass_case']}条</font></strong><br>
|
||||
失败用例个数（failed）: <strong><font color="red" >{pytest_result['fail_case']}条</font></strong><br>
|
||||
报错用例个数（error）: <strong><font color="orange" >{pytest_result['error_case']}条</font></strong><br>
|
||||
跳过用例个数（skipped）: <strong><font color="grey" >{pytest_result['skip_case']}条</font></strong><br>
|
||||
预期失败用例个数（xfail）: <strong><font color="grey" >{pytest_result['xfail_case']}条</font></strong><br>
|
||||
预期通过用例个数（xpass）: <strong><font color="grey" >{pytest_result['xpass_case']}条</font></strong><br>
|
||||
通过率: <strong><font color="green" >{pytest_result['pass_rate']}%</font></strong><br>
|
||||
测试报告，点击查看: <a href='{ALLURE_URL}'>[测试报告入口]</a><br>
|
||||
构建详情，点击查看: <a href='{BUILD_URL}'>[控制台入口]</a><br>
|
||||
用例运行总数:<strong> ${total_case}条</strong><br>
|
||||
通过用例数（passed）: <strong><font color="green" >${pass_case}条</font></strong><br>
|
||||
失败用例数（failed）: <strong><font color="red" >${fail_case}条</font></strong><br>
|
||||
报错用例数（error）: <strong><font color="orange" >${error_case}条</font></strong><br>
|
||||
跳过用例数（skipped）: <strong><font color="grey" >${skip_case}条</font></strong><br>
|
||||
预期失败用例数（xfail）: <strong><font color="grey" >${xfail_case}条</font></strong><br>
|
||||
预期通过用例数（xpass）: <strong><font color="grey" >${xpass_case}条</font></strong><br>
|
||||
通过率: <strong><font color="green" >${pass_rate}%%</font></strong><br>
|
||||
测试报告，点击查看: <a href='%s'>[测试报告入口]</a><br>
|
||||
构建详情，点击查看: <a href='%s'>[控制台入口]</a><br>
|
||||
|
||||
**********************************<br>
|
||||
附件为具体的测试报告，详细情况可下载附件查看， 非相关负责人员可忽略此消息。谢谢。
|
||||
"""
|
||||
""" % (ProjectName, BUILD_NUMBER, ENV.name, ALLURE_URL, BUILD_URL)
|
||||
|
||||
# ------------------------------------ 企业微信相关配置 ----------------------------------------------------#
|
||||
|
||||
# 企业微信通知群聊
|
||||
wechat_webhook_url = ["https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxxxx"]
|
||||
|
||||
# 企业微信通知内容
|
||||
wechat_content = f"""******用例执行结果统计******
|
||||
> 项目名称:{ProjectName}
|
||||
> 构件编号:#{BUILD_NUMBER}
|
||||
> 测试环境:{ENV.name}
|
||||
> 总用例数：<font color=\"info\">{pytest_result['total_case']}条</font>
|
||||
> 通过用例数：<font color=\"info\">{pytest_result['pass_case']}条</font>
|
||||
> 失败用例数：<font color=\"red\">{pytest_result['fail_case']}条</font>
|
||||
> 报错用例数：<font color=\"red\">{pytest_result['error_case']}条</font>
|
||||
> 跳过用例数：<font color=\"warning\">{pytest_result['skip_case']}条</font>
|
||||
> 预期失败用例数：<font color=\"comment\">{pytest_result['xfail_case']}条</font>
|
||||
> 预期通过用例数：<font color=\"comment\">{pytest_result['xpass_case']}条</font>
|
||||
> 通过率：<font color=\"info\">{pytest_result['pass_rate']}%</font>
|
||||
> 用例开始时间:<font color=\"info\">{pytest_result["case_start_time"]}</font>
|
||||
> 用例执行时长：<font color=\"info\">{pytest_result['case_duration']}s</font>
|
||||
> 测试报告，点击查看>>[测试报告入口]({ALLURE_URL})
|
||||
> 构建详情，点击查看>>[控制台入口]({BUILD_URL})
|
||||
> <@汪杰>"""
|
||||
wechat_content = """******用例执行结果统计******
|
||||
> 项目名称:$%s
|
||||
> 构件编号:#%s
|
||||
> 测试环境:$%s
|
||||
> 总用例数：<font color=\"info\">${total_case}条</font>
|
||||
> 通过用例数：<font color=\"info\">${pass_case}条</font>
|
||||
> 失败用例数：<font color=\"red\">${fail_case}条</font>
|
||||
> 报错用例数：<font color=\"red\">${error_case}条</font>
|
||||
> 跳过用例数：<font color=\"warning\">${skip_case}条</font>
|
||||
> 预期失败用例数：<font color=\"comment\">${xfail_case}条</font>
|
||||
> 预期通过用例数：<font color=\"comment\">${xpass_case}条</font>
|
||||
> 通过率：<font color=\"info\">${pass_rate}%%</font>
|
||||
> 用例开始时间:<font color=\"info\">${case_start_time}</font>
|
||||
> 用例执行时长：<font color=\"info\">${case_duration}s</font>
|
||||
> 测试报告，点击查看>>[测试报告入口](%s)
|
||||
> 构建详情，点击查看>>[控制台入口](%s)
|
||||
> <@汪杰>""" % (ProjectName, BUILD_NUMBER, ENV.name, ALLURE_URL, BUILD_URL)
|
||||
|
|
|
|||
24
run.py
24
run.py
|
|
@ -20,7 +20,9 @@ from common.robot_sender import EnterpriseWechatNotification
|
|||
from common.settings import IS_SEND_EMAIL, IS_SEND_WECHAT, wechat_webhook_url, wechat_content, email_content, \
|
||||
email_config
|
||||
from configs.dir_path_config import BASE_DIR, TEMP_DIR, PYTEST_REPORT_DIR, PYTEST_RESULT_DIR, ALLURE_REPORT_DIR
|
||||
from utils.data_handle import DataProcessor
|
||||
from utils.file_handle import FileHandle
|
||||
from utils.reportdatahandle import ReportDataHandle
|
||||
|
||||
if __name__ == '__main__':
|
||||
logger.info("""
|
||||
|
|
@ -66,23 +68,13 @@ if __name__ == '__main__':
|
|||
FileHandle.copy_file(BASE_DIR + os.sep + '查看allure报告方法', ALLURE_REPORT_DIR)
|
||||
|
||||
# 发送企业微信群聊
|
||||
pytest_result = ReportDataHandle.pytest_json_report_case_count()
|
||||
if IS_SEND_WECHAT: # 判断是否需要发送企业微信
|
||||
EnterpriseWechatNotification(wechat_webhook_url).send_markdown(wechat_content)
|
||||
EnterpriseWechatNotification(wechat_webhook_url).send_markdown(
|
||||
DataProcessor().process_data(wechat_content, pytest_result))
|
||||
|
||||
# 发送邮件
|
||||
# 发送邮件
|
||||
if IS_SEND_EMAIL: # 判断是否需要发送邮件
|
||||
file_path = PYTEST_REPORT_DIR + os.sep + 'pytest_report.html'
|
||||
# with open(file_path, 'rb') as f:
|
||||
# text_to_send = f.read()
|
||||
|
||||
# config = YamlHandle(CONFIGS_DIR + os.sep + 'mail_config.yaml').read_yaml()
|
||||
config = email_config
|
||||
ms = MailSender(
|
||||
mail_subject=config['mail_subject'],
|
||||
sender_username=config['sender_username'],
|
||||
sender_password=config['sender_password'],
|
||||
receiver_mail_list=config['receiver_mail_list'],
|
||||
smtp_domain=config['smtp_domain'],
|
||||
smtp_port=config['smtp_port'],
|
||||
)
|
||||
ms.attach_text(email_content).attach_file(file_path).send()
|
||||
ms = MailSender(**email_config)
|
||||
ms.attach_text(DataProcessor().process_data(email_content, pytest_result)).attach_file(file_path).send()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,97 @@
|
|||
# !/usr/bin/python
|
||||
# -*- coding:utf-8 -*-
|
||||
# @Time : 2023/9/12 16:36
|
||||
# @Author : wangjie
|
||||
# @File : data_handle.py
|
||||
# @project : SensoroApiAutoTest
|
||||
import ast
|
||||
import re
|
||||
from string import Template
|
||||
from typing import Any
|
||||
|
||||
from common.exceptions import DataProcessorFuncError
|
||||
from utils.faker_utils import FakerUtils
|
||||
from utils.time_utils import TimeUtil
|
||||
|
||||
|
||||
class DataProcessor:
|
||||
|
||||
def process_data(self, data: Any, source=None) -> Any:
|
||||
"""
|
||||
处理输入的数据，根据数据类型分发到不同的处理方法中。
|
||||
:param data: 输入的数据，可以是字符串、列表或字典。
|
||||
:param source: 数据源，用于字符串模板替换。
|
||||
:return: 处理后的数据
|
||||
"""
|
||||
if source is None:
|
||||
source = {}
|
||||
data = self.eval_data(data)
|
||||
if isinstance(data, str):
|
||||
data = self.process_string(data, source)
|
||||
elif isinstance(data, list):
|
||||
data = [self.process_data(item, source) for item in data]
|
||||
elif isinstance(data, dict):
|
||||
data = {key: self.process_data(value, source) for key, value in data.items()}
|
||||
return data
|
||||
|
||||
def eval_data(self, data: Any) -> Any:
|
||||
"""
|
||||
尝试解析输入的数据，使用ast.literal_eval模块，只会执行合法的Python表达式，例如将："[1,2,3]" 或者"{'k':'v'}" -> [1,2,3], {'k':'v'}可以执行，'1+1'这种不会执行
|
||||
:param data: 输入的数据，通常是一个字符串。
|
||||
:return: 解析后的数据，如果无法解析则返回原始输入数据。
|
||||
"""
|
||||
try:
|
||||
return ast.literal_eval(data)
|
||||
except (SyntaxError, ValueError, TypeError):
|
||||
return data
|
||||
|
||||
def process_string(self, s: str, source: dict) -> str:
|
||||
"""
|
||||
处理输入的字符串，包括变量替换和函数执行。
|
||||
:param s: 输入的字符串，可能包含变量和函数调用。
|
||||
:param source: 数据源，用于字符串模板替换。
|
||||
:return: 处理后的字符串
|
||||
"""
|
||||
s = Template(s).safe_substitute(source)
|
||||
for func in re.findall('\\${(.*?)}', s):
|
||||
try:
|
||||
func_result = eval(func)
|
||||
s = s.replace('${%s}' % func, repr(func_result) if isinstance(func_result, str) else str(func_result))
|
||||
except NameError:
|
||||
# 处理未定义的变量或函数
|
||||
raise DataProcessorFuncError(
|
||||
f'方法执行错误，请检查data_handle.py中是否导入该方法，或者方法名称是否正确，方法名：{func}')
|
||||
except Exception as e:
|
||||
# 处理其他异常情况
|
||||
raise DataProcessorFuncError(
|
||||
f'方法执行错误：{func}, 报错信息：{e}')
|
||||
|
||||
return self.eval_data(s)
|
||||
|
||||
|
||||
# 示例用法
|
||||
if __name__ == "__main__":
|
||||
data_processor = DataProcessor()
|
||||
source_data = {
|
||||
"name": "John",
|
||||
"age": 30,
|
||||
"random_int": FakerUtils().random_int()
|
||||
}
|
||||
input_data = {
|
||||
"message": "Hello, ${FakerUtils().random_name()}! Your age is ${age}. Random number: ${FakerUtils().random_int()}.",
|
||||
"nested_data": [
|
||||
"This is ${name}'s data.",
|
||||
{
|
||||
"message": "Age: ${age}.",
|
||||
"nested_list": [
|
||||
"More data: ${random_int}.",
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
# input_data = '[[1,2,3,4],${FakerUtils().random_name()}]'
|
||||
# input_data = '${FakerUtils().random_name()}'
|
||||
# input_data = "Hello, ${FakerUtils().random_name()}! Your age is ${age}. Random number: ${FakerUtils().random_int()}."
|
||||
# input_data = [[1, 2, 3, 4], '${FakerUtils().random_name()}']
|
||||
processed_data = data_processor.process_data(input_data, source_data)
|
||||
print(processed_data, type(processed_data))
|
||||
|
|
@ -9,7 +9,7 @@ import random
|
|||
from faker import Faker
|
||||
|
||||
|
||||
class FakerUtiles:
|
||||
class FakerUtils:
|
||||
|
||||
def __init__(self):
|
||||
self.faker = Faker(locale='zh_CN')
|
||||
|
|
@ -75,6 +75,6 @@ class FakerUtiles:
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
s = FakerUtiles().random_IDcard()
|
||||
s = FakerUtils().random_IDcard()
|
||||
print(s)
|
||||
print(type(s))
|
||||
|
|
|
|||
Loading…
Reference in New Issue