feat: 重构前端项目结构并优化代码

refactor: 迁移前端资源文件至web目录
feat: 新增多种图标资源
style: 统一代码风格和格式化配置
docs: 更新README和文档说明
chore: 更新依赖和配置文件
fix: 修复部分类型定义和枚举
perf: 优化路由和组件加载逻辑
This commit is contained in:
zhangtao
2026-04-29 00:02:09 +08:00
parent 10a35bb4b5
commit eb150bcede
887 changed files with 116514 additions and 33990 deletions
View File
+101
View File
@@ -0,0 +1,101 @@
# -*- coding: utf-8 -*-#
# 钉钉发送消息模块
from typing import Any
from dingtalkchatbot.chatbot import DingtalkChatbot, FeedLink
class DingTalkPack:
def __init__(self, webhook: str, secret: str, at_mobiles: list = None):
self.webhook=webhook
self.secret=secret
self.at_mobiles=at_mobiles
self.ding_news = DingtalkChatbot(webhook=self.webhook, secret=self.secret, pc_slide=False, fail_notice=False)
def send_text(self, msg: str, mobiles: list = None) -> None:
"""
发送文本信息
:param msg: 文本内容
:param mobiles: 用户电话
:return:
"""
if not mobiles:
self.ding_news.send_text(msg=msg, is_at_all=True)
else:
if isinstance(mobiles, list):
self.ding_news.send_text(msg=msg, at_mobiles=mobiles)
else:
raise TypeError("mobiles类型错误 不是list类型.")
def send_link(self, title: str, text: str, message_url: str, pic_url: str) -> None:
"""
发送link通知
:return:
"""
try:
self.ding_news.send_link(title=title, text=text, message_url=message_url, pic_url=pic_url)
except Exception:
raise
def send_markdown(self, title: str, msg: str, mobiles: list = None) -> None:
"""
:param mobiles:
:param title:
:param msg:
markdown 格式
"""
if mobiles is None:
self.ding_news.send_markdown(title=title, text=msg, is_at_all=True)
else:
if isinstance(mobiles, list):
self.ding_news.send_markdown(title=title, text=msg, at_mobiles=mobiles)
else:
raise TypeError("mobiles类型错误 不是list类型.")
def feed_link(self, title: str, message_url: str, pic_url: str) -> Any:
"""
发送link类型
:param title:
:param message_url:
:param pic_url:
:return:
"""
return FeedLink(title=title, message_url=message_url, pic_url=pic_url)
def send_feed_link(self, *arg) -> None:
"""
发送feedlink类型
:param arg:
:return:
"""
try:
self.ding_news.send_feed_card(list(arg))
except Exception:
raise
def send_dingding(self, title, environment, tester, total, pass_num, fail_num, error_num, skip_num, rate, duration, reprot_url, jenkins_url):
"""
发送钉钉通知
:return:
"""
self.ding_news.send_markdown(
title=f'【{title}测试执行完毕提醒',
text=f"<font color=\'#FFA500\'>[通知] </font>测试执行完成 \n\n --- \n\n" +
"执行环境:<font color=\"#1d953f\">%s </font>" % environment + "\n\n" +
"执行人员:<font color=\"#f2eada\">@%s</font>" % tester + "\n\n --- \n\n" +
"运行总数:<font color=\"#d71345\">%s </font>" % total + "\n\n" +
"通过数量:<font color=\"#1d953f\">%s </font>" % pass_num + "\n\n" +
"失败数量:<font color=\"#c63c26\">%s </font>" % fail_num + "\n\n" +
"异常数量:<font color=\"#fdb933\">%s </font>" % error_num + "\n\n" +
"跳过数量:<font color=\"#2585a6\">%s </font>" % skip_num + "\n\n" +
"成功率为:<font color=\"#1d953f\">%s </font>" % rate + "\n\n" +
"运行时间:<font color=\"#464547\">%s </font>" % duration + "\n\n" +
"报告详情:[点击查看](%s)" % reprot_url + "\n\n" +
"构建地址:[点击查看](%s) 详细情况可登录jenkins平台查看,非相关负责人员可忽略此消息。谢谢。" % jenkins_url ,
at_mobiles=self.at_mobiles
)
+89
View File
@@ -0,0 +1,89 @@
# -*- coding: utf-8 -*-
import smtplib
import time
from email.header import Header
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import logging
class EmailPack:
# 初始化发件人,密码,收件人列表
def __init__(self, fromaddr: str, password: str, toaddrs: list, server_host: str):
"""
:param REPORT_END_PATH:
"""
self.fromaddr = fromaddr
self.password = password
self.toaddrs = toaddrs
self.server_host = server_host
self.server = smtplib.SMTP(self.server_host)
self.message = MIMEMultipart()
# 设置发件人名称,主题,内容,附件
def _set_message(self, name: str, title: str, content: str, filelist: list):
"""
:param name:
:param title:
:param content:
:param filelist:
:return:
"""
tm = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
self.message['From'] = Header(f"{name}<{self.fromaddr}>", 'utf-8') # 发件人名称和地址
self.message['Subject'] = Header(title + "_" + tm, 'utf-8') # 邮件主题
self.message.attach(MIMEText(content)) # 邮件内容
if filelist is not None: # 邮件附件
for file in filelist:
fileApart = MIMEApplication( open(file, 'rb').read(), file.split('.')[-1])
fileApart.add_header('Content-Disposition', 'attachment', filename=file.split("\\")[-1])
self.message.attach(fileApart)
# 发送邮件
def _send_message(self):
"""
:return:
"""
try:
self.server.login(self.fromaddr, self.password)
self.server.sendmail(self.fromaddr, self.toaddrs, self.message.as_string())
logging.info(f'【邮件发送成功!收件人:{self.toaddrs}】')
self.server.quit()
except smtplib.SMTPException as e:
logging.error(f'【邮件发送异常!{e}】')
# 默认发送邮件
def send_default_email(self, title: str, environment: str, tester: str, total: int, pass_num: int, fail_num: int, error_num: int, skip_num: int, rate: str, duration: str, reprot_url: str, jenkins_url: str, report_path: list):
"""
:rtype: object
:return:
"""
self._set_message(
name="自动化测试",
title=f'{title}测试执行完毕提醒!',
content=f'''
各位同事, 大家好:
自动化用例执行完成,执行结果如下:
**********************************
执行环境: {environment}
执行人员: {tester}
运行总数: {total}
通过: {pass_num}
失败: {fail_num}
异常: {error_num}
跳过: {skip_num}
成功率: {rate}
总耗时: {duration}
报告详情: [点击查看]({reprot_url})
**********************************
jenkins地址:{jenkins_url}
详细情况可登录jenkins平台查看,非相关负责人员可忽略此消息。谢谢。
''',
filelist=[report_path]
)
self._send_message()
+118
View File
@@ -0,0 +1,118 @@
# -*- coding: utf-8 -*-
import json
from typing import Dict, List, Optional
import pandas as pd
from openpyxl.styles import Font
from openpyxl import load_workbook
import logging
class ExcelPack:
# Excel列结构配置
COLUMNS = {
'name': 0,
'run': 1,
'token': 2,
'url': 3,
'data': 4,
'method': 5,
'type': 6,
'header': 7,
'code': 8,
'isSuccess': 9,
'result': 10,
}
# 结果样式配置
STYLES = {
'pass': (3, '00FF00'), # 绿色
'fail': (4, 'FF0000'), # 红色
'skip': (6, '800080'), # 紫色
}
def __init__(self, file_name: str, api_host: str, token: str):
self.file_name = file_name
self._df = None
self.api_host = api_host
self.token = token
@property
def df(self) -> pd.DataFrame:
if self._df is None:
self._df = pd.read_excel(self.file_name)
logging.info(f"成功读取Excel文件: {self.file_name}")
return self._df
def _get_cell_value(self, row: int, col_name: str) -> str:
"""统一的单元格获取方法"""
value = self.df.iloc[row, self.COLUMNS[col_name]]
return str(value).strip() if pd.notna(value) else ''
def _write_cell(self, row: int, col_name: str, value: str, style: Optional[int] = None):
"""增量更新单元格内容,保留其他数据格式"""
# 更新DataFrame中的值
col = self.COLUMNS[col_name]
self.df.iloc[row, col] = value
# 直接加载工作簿进行更新
workbook = load_workbook(self.file_name)
worksheet = workbook.active
# 写入新值
cell = worksheet.cell(row=row + 2, column=col + 1)
cell.value = value
# 应用样式
if style and isinstance(style, (int, str)):
style_config = self.STYLES.get(str(style))
if style_config and len(style_config) >= 2:
color = style_config[1]
cell.font = Font(bold=True, color=color)
else:
logging.warning(f"无效的样式配置: {style}")
# 保存工作簿
workbook.save(self.file_name)
def load_test_cases(self) -> List[Dict]:
"""加载测试用例"""
test_cases = []
for row in range(1, len(self.df)):
headers = eval(self._get_cell_value(row, 'header')) if self._get_cell_value(row, 'header') else {}
if self._get_cell_value(row, 'token').lower() == 'yes':
headers['token'] = self.token
case = {
'name': self._get_cell_value(row, 'name'),
'run': self._get_cell_value(row, 'run'),
'token': self._get_cell_value(row, 'token'),
'type': self._get_cell_value(row, 'type'),
'request': {
'url': f"{self.api_host}{self._get_cell_value(row, 'url')}",
'method': self._get_cell_value(row, 'method'),
'headers': headers,
'data': json.loads(self._get_cell_value(row, 'data')) if self._get_cell_value(row, 'data') else None,
},
'expected': {
'code': int(self._get_cell_value(row, 'code')),
'isSuccess': bool(self._get_cell_value(row, 'isSuccess')),
},
'result': {
'result': self._get_cell_value(row, 'result'),
},
'row': row,
}
test_cases.append(case)
return test_cases
def write_to_excel(self, row: int, result: Dict) -> None:
"""写入结果到Excel"""
try:
# 更新数据
self._write_cell(row, 'result', result['result'], result['result'])
logging.info(f"测试结果写入成功: {result['name']}")
except Exception as e:
logging.error(f"写入测试结果失败: {str(e)}")
logging.exception(e)
+56
View File
@@ -0,0 +1,56 @@
# -*- coding: utf-8 -*-
import logging
from typing import Dict
from common.reponse_tool import Response
from common.request_tool import Requests
class Manage:
def __init__(self):
pass
def create_result(self, name: str, status: str = 'skip', response: str = None) -> Dict:
"""创建统一的结果对象"""
return {
'name': name,
'result': status,
'response': response
}
def execute_test_case(self, case: Dict) -> Dict:
"""执行测试用例"""
try:
# 构建请求数据
request_data = {
'url': case['request']['url'],
'method': case['request']['method'],
'data': case['request']['data'],
'headers': case['request']['headers'],
'data_type': case.get('type', 'json')
}
# 发送请求并获取响应
response = Response().result(Requests().send_request(**request_data))
# 断言
# 情况1:如果响应体包含'isSuccess'字段(按预期格式)
if case['expected']['isSuccess'] and case['expected']['code'] == response['code']:
status = 'pass'
else:
status = 'fail'
result = self.create_result(case['name'], status, response)
logging.info(f"用例执行完成: {case['name']}, 结果: {status}")
except Exception as e:
error_msg = f"执行用例异常: {str(e)}"
logging.error(error_msg)
result = self.create_result(case['name'], 'fail', error_msg)
return result
+182
View File
@@ -0,0 +1,182 @@
# -*- coding: utf-8 -*-#
# 企业微信发送消息模块
import requests
import json
from typing import Any, List, Optional
class QiWeiPack:
"""
企业微信机器人消息发送封装类
"""
def __init__(self, webhook_url: str):
"""
初始化企业微信机器人
:param webhook_url: 企业微信机器人的webhook地址
"""
self.webhook_url = webhook_url
self.headers = {"Content-Type": "application/json"}
def _send_request(self, data: dict) -> dict:
"""
发送请求到企业微信API
:param data: 消息数据
:return: 响应结果
"""
try:
response = requests.post(
self.webhook_url,
data=json.dumps(data),
headers=self.headers
)
response.raise_for_status()
return response.json()
except Exception as e:
raise Exception(f"发送企业微信消息失败: {str(e)}")
def send_text(self, content: str, mentioned_list: Optional[List[str]] = None, mentioned_mobile_list: Optional[List[str]] = None) -> dict:
"""
发送文本消息
:param content: 文本内容
:param mentioned_list: @的用户列表,如["user1", "user2"]
:param mentioned_mobile_list: @的手机号列表,如["13800138000"]
:return: 响应结果
"""
data = {
"msgtype": "text",
"text": {
"content": content
}
}
if mentioned_list:
data["text"]["mentioned_list"] = mentioned_list
if mentioned_mobile_list:
data["text"]["mentioned_mobile_list"] = mentioned_mobile_list
return self._send_request(data)
def send_markdown(self, content: str) -> dict:
"""
发送markdown消息
:param content: markdown内容
:return: 响应结果
"""
data = {
"msgtype": "markdown",
"markdown": {
"content": content
}
}
return self._send_request(data)
def send_image(self, base64: str, md5: str) -> dict:
"""
发送图片消息
:param base64: 图片base64编码
:param md5: 图片md5值
:return: 响应结果
"""
data = {
"msgtype": "image",
"image": {
"base64": base64,
"md5": md5
}
}
return self._send_request(data)
def send_news(self, articles: List[dict]) -> dict:
"""
发送图文消息
:param articles: 图文列表,每个元素包含title、description、url、picurl字段
:return: 响应结果
"""
data = {
"msgtype": "news",
"news": {
"articles": articles
}
}
return self._send_request(data)
def send_file(self, media_id: str) -> dict:
"""
发送文件消息
:param media_id: 文件媒体ID(需要先上传文件获取)
:return: 响应结果
"""
data = {
"msgtype": "file",
"file": {
"media_id": media_id
}
}
return self._send_request(data)
def send_taskcard(self, title: str, description: str, url: str, btn_json_list: List[dict]) -> dict:
"""
发送任务卡片消息
:param title: 标题
:param description: 描述
:param url: 点击卡片跳转的URL
:param btn_json_list: 按钮列表
:return: 响应结果
"""
data = {
"msgtype": "taskcard",
"taskcard": {
"title": title,
"description": description,
"url": url,
"btn_json_list": btn_json_list
}
}
return self._send_request(data)
def send_test_report(self, title: str, environment: str, tester: str, total: int, pass_num: int,
fail_num: int, error_num: int, skip_num: int, rate: str, duration: str,
report_url: str, jenkins_url: str) -> dict:
"""
发送测试报告通知
:param title: 测试标题
:param environment: 执行环境
:param tester: 执行人员
:param total: 运行总数
:param pass_num: 通过数量
:param fail_num: 失败数量
:param error_num: 异常数量
:param skip_num: 跳过数量
:param rate: 成功率
:param duration: 运行时间
:param report_url: 报告地址
:param jenkins_url: Jenkins地址
:return: 响应结果
"""
# 企业微信markdown支持的颜色有限,这里使用emoji和加粗来突出显示
content = (
f"## {title}测试执行完毕提醒\n\n" +
"📢 **测试执行完成**\n\n" +
"---\n\n" +
f"**执行环境:** {environment}\n\n" +
f"**执行人员:** {tester}\n\n" +
"---\n\n" +
f"**运行总数:** {total}\n\n" +
f"**通过数量:** ✅ {pass_num}\n\n" +
f"**失败数量:** ❌ {fail_num}\n\n" +
f"**异常数量:** ⚠️ {error_num}\n\n" +
f"**跳过数量:** ⏭️ {skip_num}\n\n" +
f"**成功率:** {rate}\n\n" +
f"**运行时间:** {duration}\n\n" +
f"**报告详情:** [点击查看]({report_url})\n\n" +
f"**构建地址:** [点击查看]({jenkins_url})\n\n" +
"详细情况可登录Jenkins平台查看,非相关负责人员可忽略此消息。谢谢。"
)
return self.send_markdown(content)
+281
View File
@@ -0,0 +1,281 @@
# -*- coding: utf-8 -*-
import string
import random
import datetime
from faker import Faker
from dateutil.relativedelta import relativedelta
class ContextPack:
def __init__(self):
self.faker = Faker(locale='zh_CN')
@property
def get_phone(self) -> int:
"""
:return: 随机生成手机号码
"""
phone = self.faker.phone_number()
return phone
@property
def get_id_number(self) -> int:
"""
:return: 随机生成身份证号码
"""
id_number = self.faker.ssn()
return id_number
@property
def get_female_name(self) -> str:
"""
:return: 女生姓名
"""
female_name = self.faker.name_male()
return female_name
@property
def get_male_name(self) -> str:
"""
:return: 男生姓名
"""
male_name = self.faker.name_female()
return male_name
@property
def get_email(self) -> str:
"""
:return: 生成邮箱
"""
email = self.faker.email()
return email
@property
def merchantSelfOperatedShop(self) -> int:
"""
:return: 商家端自营店铺ID
"""
SelfOperatedShop = 515
return SelfOperatedShop
@property
def get_secend_time(self)->datetime:
"""
计算当前时间: 年-月-日 时:分:秒:毫秒
:return:
"""
return datetime.datetime.now()
@property
def get_day_time(self)->str:
"""
计算当前时间: 年-月-日
:return:
"""
return datetime.datetime.now().strftime('%Y-%m-%d')
def generate_bank_card(self):
"""
生成随机银行卡号
"""
return self.faker.credit_card_number()
def generate_email(self):
"""
生成随机邮箱号
"""
return self.random_str(4) + self.faker.free_email()
@property
def generate_company(self):
"""
生成随机公司名称
"""
return self.faker.company_prefix() + self.random_str(5) + '测试' + self.faker.company_suffix()
@property
def generate_name(self):
"""
生成随机姓名
"""
return self.faker.name()
@property
def generate_unreal_phone(self):
"""
生成不存在的手机号
"""
phone = random.choice(['100', '110', '120']) + \
''.join(random.choice('0123456789') for _ in range(8))
return phone
def random_str(self, str_len: int) -> str:
"""从a-zA-Z0-9生成制定数量的随机字符
:param str_len: 字符串长度
:return:
"""
try:
str_len = int(str_len)
except ValueError:
raise Exception("调用随机字符失败,[ %s ]长度参数有误!" % str_len)
strings = ''.join(random.sample(string.hexdigits, +str_len))
return strings
def random_int(self, scope) -> int:
"""获取随机整型数据
:param scope: 数据范围
:return:
"""
try:
start_num, end_num = scope.split(",")
start_num = int(start_num)
end_num = int(end_num)
except ValueError:
raise Exception("调用随机整数失败,[ %s ]范围参数有误!" % str(scope))
if start_num <= end_num:
number = random.randint(start_num, end_num)
else:
number = random.randint(end_num, start_num)
return number
def random_float(self, data) -> float:
"""获取随机浮点数据
:param data: 数组
:return:
"""
try:
start_num, end_num, accuracy = data.split(",")
start_num = int(start_num)
end_num = int(end_num)
accuracy = int(accuracy)
except ValueError:
raise Exception("调用随机浮点数失败,[ %s ]范围参数或精度有误!" % data)
if start_num <= end_num:
number = random.uniform(start_num, end_num)
else:
number = random.uniform(end_num, start_num)
number = round(number, accuracy)
return number
def random_choice(self, data):
"""获取数组随机值
:param data: 数组
:return:
"""
_list = data.split(",")
each = random.choice(_list)
return each
def get_date_mark(self, now, mark, num):
if 'y' == mark:
return now + relativedelta(years=num)
elif 'm' == mark:
return now + relativedelta(months=num)
elif 'd' == mark:
return now + relativedelta(days=num)
elif 'h' == mark:
return now + relativedelta(hours=num)
elif 'M' == mark:
return now + relativedelta(minutes=num)
elif 's' == mark:
return now + relativedelta(seconds=num)
else:
raise Exception("日期字段标识[ %s ]错误, 请使用[年y,月m,日d,时h,分M,秒s]标识!" % mark)
def generate_date(self, expr=''):
"""生成日期对象(不含时分秒)
:param expr: 日期表达式,如"d-1"代表日期减1
:return:
"""
today = datetime.date.today()
if expr:
try:
mark = expr[:1]
num = int(expr[1:])
except (TypeError, NameError):
raise Exception("调用生成日期失败,日期表达式[ %s ]有误!" % expr)
return self.get_date_mark(today, mark, num)
else:
return today
def generate_datetime(self, expr=''):
"""生成日期时间对象(含时分秒)
:param expr: 日期表达式,如"d-1"代表日期减1
:return:
"""
now = datetime.datetime.now().replace(microsecond=0)
if expr:
try:
mark = expr[:1]
num = int(expr[1:])
except (TypeError, NameError):
raise Exception("调用生成日期失败,日期表达式[ %s ]有误!" % expr)
return self.get_date_mark(now, mark, num)
else:
return now
def generate_timestamp(self, expr='') -> int:
"""生成时间戳(13位)
:param expr: 日期表达式,如"d-1"代表日期减1
:return:
"""
datetime_obj = self.generate_datetime(expr)
return int(datetime.datetime.timestamp(datetime_obj)) * 1000
def generate_guid(self) -> str:
"""基于MAC地址+时间戳+随机数来生成GUID
:param:
:return:
"""
import uuid
return str(uuid.uuid1()).upper()
def generate_wxid(self):
"""基于AUTO标识+26位英文字母大小写+数字生成伪微信ID
:param:
:return:
"""
return 'AUTO' + ''.join(random.sample(string.ascii_letters + string.digits, 24))
def generate_noid(self, expr=''):
"""基于6位随机数字+出生日期+4位随机数生成伪身份证
:param expr: 日期表达式,如"d-1"代表日期减1
:return:
"""
# birthday = generate_date(expr)
# birthday = str(birthday).replace('-', '')
# return int(str(random.randint(100000, 999999)) + birthday + str(random.randint(1000, 9999)))
faker = Faker(locale='zh_CN')
return faker.ssn()
def generate_phone(self):
"""基于三大运营商号段+随机数生成伪手机号
:param:
:return:
"""
ctcc = [133, 153, 173, 177, 180, 181, 189, 191, 193, 199]
cucc = [130, 131, 132, 155, 156, 166, 175, 176, 185, 186, 166]
cmcc = [134, 135, 136, 137, 138, 139, 147, 150, 151, 152, 157, 158, 159, 172, 178, 182, 183, 184, 187, 188, 198]
begin = 10 ** 7
end = 10 ** 8 - 1
prefix = random.choice(ctcc + cucc + cmcc)
return str(prefix) + str(random.randint(begin, end))
+33
View File
@@ -0,0 +1,33 @@
# -*- coding: utf-8 -*-
from typing import Dict, List
import yaml
from string import Template
import logging
class YamlPack:
# yaml文件全部内容
def __init__(self, yaml_path, api_host: str, token: str):
# yaml中的变量
variable = {
'host': api_host,
'token': token
}
self.pass_num = 0
self.fail_num = 0
self.yaml_path = yaml_path
with open(self.yaml_path, 'r', encoding="utf-8") as f:
re = Template(f.read()).substitute(variable)
self.data = yaml.safe_load(re)
logging.info(f"成功读取Yaml文件: {self.yaml_path}")
# Test目录下的全部用例
def load_test_cases(self) -> List[Dict]:
test_list = []
case = self.data["Case"]
for a in case:
for k, v in a.items():
if k == "Test":
test_list.append(v)
return test_list