mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-20 20:39:55 +00:00
chore: 更新版本号至 3.0.0
This commit is contained in:
zhangtao
@@ -90,7 +90,7 @@
|
||||
{
|
||||
"config_name": "项目版本",
|
||||
"config_key": "sys_web_version",
|
||||
"config_value": "2.0.0",
|
||||
"config_value": "3.0.0",
|
||||
"config_type": true,
|
||||
"status": "0",
|
||||
"description": "初始化数据"
|
||||
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 245 KiB After Width: | Height: | Size: 248 KiB |
@@ -68,7 +68,7 @@ show_banner() {
|
||||
echo -e "╭──────────────────────────────────────────────╮"
|
||||
echo -e "│ \033[1;34m👋 欢迎使用 FastAPI 初始化脚本\033[0m │"
|
||||
echo -e "╰──────────────────────────────────────────────╯"
|
||||
echo "版本: 1.0.0"
|
||||
echo "版本: 3.0.0"
|
||||
echo "作者：coderxslee"
|
||||
echo ""
|
||||
}
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
# 环境配置
|
||||
TEST_TYPE=UI
|
||||
ENVIRONMENT=测试环境
|
||||
ENVIRONMENT_HOST=https://xxx.xxx.com
|
||||
ENVIRONMENT_TOKEN=xxxx
|
||||
|
||||
# 测试人员
|
||||
TESTER=张三
|
||||
|
||||
# 公共配置
|
||||
DELETE_ON_OFF=False
|
||||
EMAIL_ON_OFF=False
|
||||
DINGDING_ON_OFF=False
|
||||
QIWEI_ON_OFF=True
|
||||
JENKINS_URL=http://xxx.xxx.com:10086/job/auto_test/
|
||||
REPORT_URL=http://xxx.xxx.com:10086/job/auto_test
|
||||
|
||||
# 邮件配置
|
||||
EMAIL_SMTP_SERVER=smtp.qq.com
|
||||
EMAIL_SENDER=xx@qq.com
|
||||
EMAIL_PASSWORD=xxx
|
||||
EMAIL_RECEIVER=xx@qq.com
|
||||
|
||||
# 钉钉配置
|
||||
DINGDING_SECRET=xxx
|
||||
DINGDING_WEBHOOK=https://oapi.dingtalk.com/robot/send?access_token=xxx
|
||||
DINGDING_AT_MOBILES=["xxx"]
|
||||
|
||||
# 企业微信配置
|
||||
WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx
|
||||
|
||||
# 浏览器配置
|
||||
BROWSER_TYPE=chromium
|
||||
@@ -1,11 +0,0 @@
|
||||
/.idea/*
|
||||
/.pytest_cache/
|
||||
/venv/
|
||||
.venv
|
||||
.vscode/*
|
||||
|
||||
reports/*
|
||||
logs/*
|
||||
case/api/.DS_Store
|
||||
case/.DS_Store
|
||||
.DS_Store
|
||||
@@ -1,194 +0,0 @@
|
||||
木兰宽松许可证，第2版
|
||||
|
||||
木兰宽松许可证，第2版
|
||||
|
||||
2020年1月 http://license.coscl.org.cn/MulanPSL2
|
||||
|
||||
您对“软件”的复制、使用、修改及分发受木兰宽松许可证，第2版（“本许可证”）的如下条款的约束：
|
||||
|
||||
0. 定义
|
||||
|
||||
“软件” 是指由“贡献”构成的许可在“本许可证”下的程序和相关文档的集合。
|
||||
|
||||
“贡献” 是指由任一“贡献者”许可在“本许可证”下的受版权法保护的作品。
|
||||
|
||||
“贡献者” 是指将受版权法保护的作品许可在“本许可证”下的自然人或“法人实体”。
|
||||
|
||||
“法人实体” 是指提交贡献的机构及其“关联实体”。
|
||||
|
||||
“关联实体” 是指，对“本许可证”下的行为方而言，控制、受控制或与其共同受控制的机构，此处的控制是
|
||||
指有受控方或共同受控方至少50%直接或间接的投票权、资金或其他有价证券。
|
||||
|
||||
1. 授予版权许可
|
||||
|
||||
每个“贡献者”根据“本许可证”授予您永久性的、全球性的、免费的、非独占的、不可撤销的版权许可，您可
|
||||
以复制、使用、修改、分发其“贡献”，不论修改与否。
|
||||
|
||||
2. 授予专利许可
|
||||
|
||||
每个“贡献者”根据“本许可证”授予您永久性的、全球性的、免费的、非独占的、不可撤销的（根据本条规定
|
||||
撤销除外）专利许可，供您制造、委托制造、使用、许诺销售、销售、进口其“贡献”或以其他方式转移其“贡
|
||||
献”。前述专利许可仅限于“贡献者”现在或将来拥有或控制的其“贡献”本身或其“贡献”与许可“贡献”时的“软
|
||||
件”结合而将必然会侵犯的专利权利要求，不包括对“贡献”的修改或包含“贡献”的其他结合。如果您或您的“
|
||||
关联实体”直接或间接地，就“软件”或其中的“贡献”对任何人发起专利侵权诉讼（包括反诉或交叉诉讼）或
|
||||
其他专利维权行动，指控其侵犯专利权，则“本许可证”授予您对“软件”的专利许可自您提起诉讼或发起维权
|
||||
行动之日终止。
|
||||
|
||||
3. 无商标许可
|
||||
|
||||
“本许可证”不提供对“贡献者”的商品名称、商标、服务标志或产品名称的商标许可，但您为满足第4条规定
|
||||
的声明义务而必须使用除外。
|
||||
|
||||
4. 分发限制
|
||||
|
||||
您可以在任何媒介中将“软件”以源程序形式或可执行形式重新分发，不论修改与否，但您必须向接收者提供“
|
||||
本许可证”的副本，并保留“软件”中的版权、商标、专利及免责声明。
|
||||
|
||||
5. 免责声明与责任限制
|
||||
|
||||
“软件”及其中的“贡献”在提供时不带任何明示或默示的担保。在任何情况下，“贡献者”或版权所有者不对
|
||||
任何人因使用“软件”或其中的“贡献”而引发的任何直接或间接损失承担责任，不论因何种原因导致或者基于
|
||||
何种法律理论，即使其曾被建议有此种损失的可能性。
|
||||
|
||||
6. 语言
|
||||
|
||||
“本许可证”以中英文双语表述，中英文版本具有同等法律效力。如果中英文版本存在任何冲突不一致，以中文
|
||||
版为准。
|
||||
|
||||
条款结束
|
||||
|
||||
如何将木兰宽松许可证，第2版，应用到您的软件
|
||||
|
||||
如果您希望将木兰宽松许可证，第2版，应用到您的新软件，为了方便接收者查阅，建议您完成如下三步：
|
||||
|
||||
1， 请您补充如下声明中的空白，包括软件名、软件的首次发表年份以及您作为版权人的名字；
|
||||
|
||||
2， 请您在软件包的一级目录下创建以“LICENSE”为名的文件，将整个许可证文本放入该文件中；
|
||||
|
||||
3， 请将如下声明文本放入每个源文件的头部注释中。
|
||||
|
||||
Copyright (c) [Year] [name of copyright holder]
|
||||
[Software Name] is licensed under Mulan PSL v2.
|
||||
You can use this software according to the terms and conditions of the Mulan
|
||||
PSL v2.
|
||||
You may obtain a copy of Mulan PSL v2 at:
|
||||
http://license.coscl.org.cn/MulanPSL2
|
||||
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY
|
||||
KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
|
||||
NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
||||
See the Mulan PSL v2 for more details.
|
||||
|
||||
Mulan Permissive Software License，Version 2
|
||||
|
||||
Mulan Permissive Software License，Version 2 (Mulan PSL v2)
|
||||
|
||||
January 2020 http://license.coscl.org.cn/MulanPSL2
|
||||
|
||||
Your reproduction, use, modification and distribution of the Software shall
|
||||
be subject to Mulan PSL v2 (this License) with the following terms and
|
||||
conditions:
|
||||
|
||||
0. Definition
|
||||
|
||||
Software means the program and related documents which are licensed under
|
||||
this License and comprise all Contribution(s).
|
||||
|
||||
Contribution means the copyrightable work licensed by a particular
|
||||
Contributor under this License.
|
||||
|
||||
Contributor means the Individual or Legal Entity who licenses its
|
||||
copyrightable work under this License.
|
||||
|
||||
Legal Entity means the entity making a Contribution and all its
|
||||
Affiliates.
|
||||
|
||||
Affiliates means entities that control, are controlled by, or are under
|
||||
common control with the acting entity under this License, ‘control’ means
|
||||
direct or indirect ownership of at least fifty percent (50%) of the voting
|
||||
power, capital or other securities of controlled or commonly controlled
|
||||
entity.
|
||||
|
||||
1. Grant of Copyright License
|
||||
|
||||
Subject to the terms and conditions of this License, each Contributor hereby
|
||||
grants to you a perpetual, worldwide, royalty-free, non-exclusive,
|
||||
irrevocable copyright license to reproduce, use, modify, or distribute its
|
||||
Contribution, with modification or not.
|
||||
|
||||
2. Grant of Patent License
|
||||
|
||||
Subject to the terms and conditions of this License, each Contributor hereby
|
||||
grants to you a perpetual, worldwide, royalty-free, non-exclusive,
|
||||
irrevocable (except for revocation under this Section) patent license to
|
||||
make, have made, use, offer for sale, sell, import or otherwise transfer its
|
||||
Contribution, where such patent license is only limited to the patent claims
|
||||
owned or controlled by such Contributor now or in future which will be
|
||||
necessarily infringed by its Contribution alone, or by combination of the
|
||||
Contribution with the Software to which the Contribution was contributed.
|
||||
The patent license shall not apply to any modification of the Contribution,
|
||||
and any other combination which includes the Contribution. If you or your
|
||||
Affiliates directly or indirectly institute patent litigation (including a
|
||||
cross claim or counterclaim in a litigation) or other patent enforcement
|
||||
activities against any individual or entity by alleging that the Software or
|
||||
any Contribution in it infringes patents, then any patent license granted to
|
||||
you under this License for the Software shall terminate as of the date such
|
||||
litigation or activity is filed or taken.
|
||||
|
||||
3. No Trademark License
|
||||
|
||||
No trademark license is granted to use the trade names, trademarks, service
|
||||
marks, or product names of Contributor, except as required to fulfill notice
|
||||
requirements in section 4.
|
||||
|
||||
4. Distribution Restriction
|
||||
|
||||
You may distribute the Software in any medium with or without modification,
|
||||
whether in source or executable forms, provided that you provide recipients
|
||||
with a copy of this License and retain copyright, patent, trademark and
|
||||
disclaimer statements in the Software.
|
||||
|
||||
5. Disclaimer of Warranty and Limitation of Liability
|
||||
|
||||
THE SOFTWARE AND CONTRIBUTION IN IT ARE PROVIDED WITHOUT WARRANTIES OF ANY
|
||||
KIND, EITHER EXPRESS OR IMPLIED. IN NO EVENT SHALL ANY CONTRIBUTOR OR
|
||||
COPYRIGHT HOLDER BE LIABLE TO YOU FOR ANY DAMAGES, INCLUDING, BUT NOT
|
||||
LIMITED TO ANY DIRECT, OR INDIRECT, SPECIAL OR CONSEQUENTIAL DAMAGES ARISING
|
||||
FROM YOUR USE OR INABILITY TO USE THE SOFTWARE OR THE CONTRIBUTION IN IT, NO
|
||||
MATTER HOW IT’S CAUSED OR BASED ON WHICH LEGAL THEORY, EVEN IF ADVISED OF
|
||||
THE POSSIBILITY OF SUCH DAMAGES.
|
||||
|
||||
6. Language
|
||||
|
||||
THIS LICENSE IS WRITTEN IN BOTH CHINESE AND ENGLISH, AND THE CHINESE VERSION
|
||||
AND ENGLISH VERSION SHALL HAVE THE SAME LEGAL EFFECT. IN THE CASE OF
|
||||
DIVERGENCE BETWEEN THE CHINESE AND ENGLISH VERSIONS, THE CHINESE VERSION
|
||||
SHALL PREVAIL.
|
||||
|
||||
END OF THE TERMS AND CONDITIONS
|
||||
|
||||
How to Apply the Mulan Permissive Software License，Version 2
|
||||
(Mulan PSL v2) to Your Software
|
||||
|
||||
To apply the Mulan PSL v2 to your work, for easy identification by
|
||||
recipients, you are suggested to complete following three steps:
|
||||
|
||||
i. Fill in the blanks in following statement, including insert your software
|
||||
name, the year of the first publication of your software, and your name
|
||||
identified as the copyright owner;
|
||||
|
||||
ii. Create a file named "LICENSE" which contains the whole context of this
|
||||
License in the first directory of your software package;
|
||||
|
||||
iii. Attach the statement to the appropriate annotated syntax at the
|
||||
beginning of each source file.
|
||||
|
||||
Copyright (c) [Year] [name of copyright holder]
|
||||
[Software Name] is licensed under Mulan PSL v2.
|
||||
You can use this software according to the terms and conditions of the Mulan
|
||||
PSL v2.
|
||||
You may obtain a copy of Mulan PSL v2 at:
|
||||
http://license.coscl.org.cn/MulanPSL2
|
||||
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY
|
||||
KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
|
||||
NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
||||
See the Mulan PSL v2 for more details.
|
||||
@@ -1,86 +0,0 @@
|
||||
# Auto_Test
|
||||
|
||||
## 项目简介
|
||||
|
||||
Auto_Test 是一个完全开源免费的自动化测试框架，旨在整合多种工具和技术，提供全面的API和UI自动化测试解决方案。通过使用pytest、playwright、locust、pytest-html、requests、Jenkins等技术，本项目能够帮助测试员高效地进行自动化测试，并生成详细的测试报告。支持通过Jenkins构建发送测试结果通知到邮件、钉钉、企微，方便团队成员及时了解测试状态。
|
||||
|
||||
## 关键特性
|
||||
|
||||
- **多技术整合**：结合Python、pytest、playwright等多种工具，实现API和UI自动化测试。
|
||||
- **详细报告**：使用pytest-html生成美观且详细的测试报告。
|
||||
- **数据处理**：利用openpyxl和pandas处理Excel数据，方便测试数据管理和结果分析。
|
||||
- **通知集成**：通过DingTalk发送测试结果通知，确保团队及时了解测试状态。
|
||||
- **性能测试**：利用locust进行性能测试，评估系统在高负载下的性能表现。
|
||||
- **数据伪造**：借助Faker生成假数据，丰富测试场景。
|
||||
- **持续集成**：通过Jenkins实现持续集成，确保项目代码始终处于良好状态。
|
||||
|
||||
## 项目结构
|
||||
|
||||
```plaintext
|
||||
AUTO_TEST
|
||||
├── case # 测试数据目录
|
||||
│ ├── api # 接口测试数据目录
|
||||
│ ├── locustfiles # locust性能测试数据目录
|
||||
│ └── ui # UI测试数据目录
|
||||
├── common # 全局公共模块
|
||||
├── config # 配置文件
|
||||
├── logs # 日志目录
|
||||
├── report # 测试报告
|
||||
├── testCase # 测试用例目录
|
||||
│ ├── api # 接口测试用例目录
|
||||
│ └── ui # UI测试用例目录
|
||||
├── tools # 工具类目录
|
||||
├── .env # 环境变量配置
|
||||
├── locust.conf # locust性能测试配置文件
|
||||
├── main.py # 测试主入口
|
||||
├── pytest.ini # pytest配置文件
|
||||
├── README.md # 项目说明文档
|
||||
└── requirements.txt # 依赖包版本文件
|
||||
```
|
||||
|
||||
## 快速开始
|
||||
|
||||
1.工具：
|
||||
|
||||
- python下载地址: <https://www.python.org/download>
|
||||
- pycharm下载地址: <https://www.jetbrains.com/pycharm>
|
||||
|
||||
2.搭建步骤
|
||||
|
||||
- 2.1拉取代码
|
||||
- git clone <https://gitee.com/fastapiadmin/AutoTest.git>
|
||||
- 查看本地和远程所有分支： git branch -a
|
||||
- 切换分支：git checkout [branch-name]
|
||||
|
||||
- 2.2创建虚拟环境：
|
||||
- python -m venv venv
|
||||
- venv/Scripts/activate
|
||||
- 回车激活环境
|
||||
|
||||
- 2.3安装项目依赖包
|
||||
- pip install -r requirements.txt
|
||||
- pip install -r requirements.txt --upgrade
|
||||
|
||||
- 2.4运行项目
|
||||
- 运行前检查config目录中Config.ini文件是否配置正确
|
||||
- python main.py
|
||||
|
||||
- 2.5ui录制脚本
|
||||
- playwright codegen
|
||||
|
||||
- 如果没有找到浏览器驱动则执行下方,一次会下载三个浏览器，chromium、firefox、webkit
|
||||
playwright install
|
||||
|
||||
- 或者指定下载浏览器
|
||||
playwright install chromium
|
||||
playwright install firefox
|
||||
playwright install webkit
|
||||
|
||||
- 性能测试
|
||||
- https://docs.locust.io/en/stable/api.html
|
||||
|
||||
## 🎨 社区交流
|
||||
|
||||
| 群组二维码 | 微信支付二维码 |
|
||||
| --- | --- |
|
||||
|  |  |
|
||||
Binary file not shown.
@@ -1,142 +0,0 @@
|
||||
Case:
|
||||
- Test:
|
||||
name: test_001_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_002_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_003_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_004_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_005_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_006_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_007_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_008_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_009_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
- Test:
|
||||
name: test_010_yaml
|
||||
run: yes
|
||||
type: json
|
||||
request:
|
||||
method: POST
|
||||
url: $host/gateway/traceplan/trackPlan/getData
|
||||
headers:
|
||||
Content-Type: application/json;charset=UTF-8
|
||||
token: $token
|
||||
data: {"staSelect":"09100101","startTime":"2025-10-14 00:00:00","endTime":"2025-10-14 00:30:00","page":1,"limit":50,"showApplication":true,"trackStatus":"4"}
|
||||
expected:
|
||||
code: 200
|
||||
isSuccess: true
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
import time
|
||||
from locust import HttpUser, task, between, User
|
||||
from locust.runners import MasterRunner
|
||||
|
||||
from locust import events
|
||||
|
||||
@events.test_start.add_listener
|
||||
def on_test_start(environment, **kwargs):
|
||||
print("开始监听")
|
||||
|
||||
@events.test_stop.add_listener
|
||||
def on_test_stop(environment, **kwargs):
|
||||
print("结束监听")
|
||||
|
||||
@events.init.add_listener
|
||||
def on_locust_init(environment, **kwargs):
|
||||
if isinstance(environment.runner, MasterRunner):
|
||||
print("我在master节点上")
|
||||
else:
|
||||
print("我在worker节点或独立节点上")
|
||||
|
||||
class QuickstartUser(HttpUser):
|
||||
# 例如，让每个用户在每次任务执行之间等待0.5到10秒：
|
||||
wait_time = between(0.5, 10)
|
||||
|
||||
# 例如，下一个用户类会睡一秒，然后两秒，再三秒，依此类推。
|
||||
last_wait_time = 0
|
||||
|
||||
def wait_time(self):
|
||||
self.last_wait_time += 1
|
||||
return self.last_wait_time
|
||||
|
||||
# @task 会有一个可选的权重参数，可以用来指定任务的执行比率。在 以下示例，任务2被选中的概率是任务1的两倍：
|
||||
@task(3)
|
||||
def task1(self):
|
||||
self.client.get("/hello")
|
||||
self.client.get("/world")
|
||||
|
||||
@task(6)
|
||||
def task2(self):
|
||||
for item_id in range(10):
|
||||
self.client.get(f"/item?id={item_id}", name="/item")
|
||||
time.sleep(1)
|
||||
|
||||
|
||||
@task
|
||||
def my_task(self):
|
||||
with self.client.get("/", catch_response=True) as response:
|
||||
if response.text != "Success":
|
||||
response.failure("Got wrong response")
|
||||
elif response.elapsed.total_seconds() > 0.5:
|
||||
response.failure("Request took too long")
|
||||
elif response.status_code == 404:
|
||||
raise RescheduleTask()
|
||||
|
||||
def on_start(self):
|
||||
self.client.post("/login", json={"username":"foo", "password":"bar"})
|
||||
|
||||
# 例如，假设我们有一个WebUser类和一个MobileUser类，我们想让MobileUser类的用户权重为1，而WebUser类的用户权重为3。
|
||||
class WebUser(User):
|
||||
weight = 3
|
||||
...
|
||||
|
||||
class MobileUser(User):
|
||||
weight = 1
|
||||
...
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 56 KiB |
@@ -1,75 +0,0 @@
|
||||
# 在最佳情况下（在 -loop 内执行小请求），单个 Locust 进程（限制在一个 CPU 核心）
|
||||
# 使用 FastHttpUser 可以每秒处理大约 16000 个请求，使用 HttpUser 可以执行 4000
|
||||
# 个请求（测试于 2021 年 M1 MacBook Pro 和 Python 3.11 上）
|
||||
|
||||
from locust import TaskSet, task, HttpUser, run_single_user, FastHttpUser
|
||||
from locust.clients import ResponseContextManager
|
||||
from locust.runners import logger
|
||||
from locust import HttpUser, task, between
|
||||
|
||||
class Task(TaskSet):
|
||||
tasks = {ThreadPage:15, write_post:1}
|
||||
@tag('thread')
|
||||
@task(3)
|
||||
def demo_api(self):
|
||||
# 基于注释中的接口信息设置请求参数
|
||||
payload = {}
|
||||
|
||||
# 设置请求头
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"token": "xxxx"
|
||||
}
|
||||
|
||||
path = "/gateway/xxx"
|
||||
logger.info(f"请求路径 {path}")
|
||||
with self.client.post(path, json=payload, headers=headers, catch_response=True) as res:
|
||||
res: ResponseContextManager
|
||||
if res.status_code != 200:
|
||||
# 输出请求的完整 URL 和状态码
|
||||
logger.error(f"请求失败. 接口地址: {res.request.url}, 状态码: {res.status_code}")
|
||||
# 输出响应文本
|
||||
logger.error(f"响应文本: {res.text}")
|
||||
# 标记请求为失败
|
||||
res.failure(res.text)
|
||||
else:
|
||||
# 检查响应内容是否包含预期的数据结构
|
||||
try:
|
||||
response_json = res.json()
|
||||
if not response_json:
|
||||
res.failure("响应体为空")
|
||||
except ValueError:
|
||||
res.failure("响应体不是有效的 JSON 格式")
|
||||
|
||||
|
||||
|
||||
def on_start(self):
|
||||
logger.info('s'*10 + '测试开始' + 's'*10)
|
||||
|
||||
def on_stop(self):
|
||||
logger.info('s'*10 + '测试结束' + 's'*10)
|
||||
|
||||
# class Test(HttpUser):
|
||||
class Test(FastHttpUser):
|
||||
wait_time = between(1, 5)
|
||||
|
||||
# 根据注释中的接口地址更新host配置
|
||||
host = 'https://xxx.com'
|
||||
tasks = [Task, ]
|
||||
|
||||
@task
|
||||
def t(self):
|
||||
with self.rest("POST", "/", json={"foo": 1}) as resp:
|
||||
if resp.js is None:
|
||||
pass # no need to do anything, already marked as failed
|
||||
elif "bar" not in resp.js:
|
||||
resp.failure(f"'bar' missing from response {resp.text}")
|
||||
elif resp.js["bar"] != 42:
|
||||
resp.failure(f"'bar' had an unexpected value: {resp.js['bar']}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
# 修复变量名错误，应该是Test类而不是test
|
||||
run_single_user(Test)
|
||||
|
||||
# 启动 Locust 测试
|
||||
# locust -f locust.py
|
||||
@@ -1,81 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Dict, Any
|
||||
from playwright.sync_api import Page, expect
|
||||
|
||||
|
||||
class BasePage:
|
||||
def __init__(self, page: Page) -> None:
|
||||
self.page = page
|
||||
|
||||
def login(self, test_data: Dict[str, Any]) -> bool:
|
||||
# 设置页面加载超时
|
||||
self.page.set_default_timeout(30000)
|
||||
|
||||
# 访问登录页面并等待页面完全加载
|
||||
self.page.goto(test_data["url"])
|
||||
self.page.wait_for_load_state("networkidle") # 等待网络活动空闲
|
||||
|
||||
# 等待用户名输入框可见且可交互
|
||||
username_input = self.page.get_by_role("textbox", name="用户名")
|
||||
username_input.wait_for(state="visible", timeout=10000)
|
||||
username_input.click()
|
||||
username_input.fill(test_data["username"])
|
||||
|
||||
# 等待密码输入框可见且可交互
|
||||
password_input = self.page.get_by_role("textbox", name="请输入登录密码")
|
||||
password_input.wait_for(state="visible", timeout=10000)
|
||||
password_input.click()
|
||||
password_input.fill(test_data["password"])
|
||||
|
||||
# 等待登录按钮可见且可交互
|
||||
login_button = self.page.get_by_role("button", name="登录")
|
||||
login_button.wait_for(state="visible", timeout=10000)
|
||||
# 使用click()方法本身会等待元素可交互，不需要单独检查enabled状态
|
||||
login_button.click()
|
||||
|
||||
# 等待登录后页面加载和弹窗出现
|
||||
self.page.wait_for_load_state("networkidle")
|
||||
|
||||
# 等待SaaS系统使用引导文本出现
|
||||
self.page.wait_for_selector("text=SaaS系统使用引导", timeout=15000)
|
||||
expect(self.page.get_by_text("SaaS系统使用引导")).to_be_visible()
|
||||
expect(self.page.get_by_text("登录成功")).to_be_visible()
|
||||
|
||||
# 等待关闭按钮可见且可交互
|
||||
close_button = self.page.get_by_role("button", name="Close")
|
||||
close_button.wait_for(state="visible", timeout=10000)
|
||||
# 使用click()方法本身会等待元素可交互，不需要单独检查enabled状态
|
||||
close_button.click()
|
||||
|
||||
# 关闭登录成功提示
|
||||
self.page.locator(".el-icon.el-notification__closeBtn > svg").click()
|
||||
|
||||
# 等待弹窗关闭
|
||||
self.page.wait_for_selector("text=SaaS系统使用引导", state="hidden", timeout=5000)
|
||||
|
||||
return True
|
||||
|
||||
def logout(self) -> bool:
|
||||
# 等待用户名显示
|
||||
expect(self.page.get_by_text("涛")).to_be_visible(timeout=20000)
|
||||
self.page.get_by_text("涛").click()
|
||||
# 等待退出登录选项显示
|
||||
self.page.wait_for_selector("text=退出登录", state="visible", timeout=10000)
|
||||
|
||||
expect(self.page.get_by_text("退出登录")).to_be_visible(timeout=10000)
|
||||
self.page.get_by_text("退出登录").click()
|
||||
# 等待确认对话框显示
|
||||
self.page.wait_for_selector("text=确定要退出登录吗？", state="visible", timeout=10000)
|
||||
|
||||
expect(self.page.get_by_text("确定要退出登录吗？")).to_be_visible(timeout=10000)
|
||||
expect(self.page.get_by_role("button", name="确认")).to_be_visible(timeout=10000)
|
||||
self.page.get_by_role("button", name="确认").click()
|
||||
# 等待退出成功提示显示
|
||||
self.page.wait_for_selector("text=退出成功", state="visible", timeout=10000)
|
||||
|
||||
expect(self.page.get_by_text("退出成功")).to_be_visible(timeout=10000)
|
||||
# 关闭退出成功提示
|
||||
self.page.get_by_role("img").nth(4).click()
|
||||
|
||||
return True
|
||||
@@ -1,34 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import logging
|
||||
import requests
|
||||
|
||||
class Response:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def result(self, res: requests.Response) -> dict:
|
||||
"""
|
||||
:param res: requests响应对象
|
||||
:return: 格式化后的响应字典
|
||||
"""
|
||||
try:
|
||||
response_dicts = dict()
|
||||
|
||||
# 响应状态码
|
||||
response_dicts['code'] = res.status_code
|
||||
|
||||
# 响应body
|
||||
response_body = res.json()
|
||||
|
||||
# 确保响应体是字典格式
|
||||
if isinstance(response_body, dict):
|
||||
response_dicts['body'] = response_body
|
||||
|
||||
# 响应秒时间
|
||||
response_dicts['time_total'] = res.elapsed.total_seconds() # 秒为单位
|
||||
|
||||
logging.info(f"【请求响应结果为: {response_dicts}】")
|
||||
return response_dicts
|
||||
except Exception as e:
|
||||
logging.error(f"响应结果处理异常：{e}")
|
||||
@@ -1,42 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import requests
|
||||
import logging
|
||||
|
||||
class Requests:
|
||||
def __init__(self):
|
||||
"""
|
||||
单例模式保证测试过程中使用的都是一个session对象
|
||||
"""
|
||||
self.session = requests.Session()
|
||||
|
||||
def send_request(self,url: str, method: str, data_type: str, headers: dict = None, data: dict = None):
|
||||
"""
|
||||
:param url: 请求url
|
||||
:param method: 请求方法
|
||||
:param data_type: 入参关键字， params(查询参数类型，明文传输，一般在url?参数名=参数值), data(一般用于form表单类型参数), json(一般用于json类型请求参数)
|
||||
:param headers: 请求头
|
||||
:param data: 参数数据，默认等于None
|
||||
:return: 返回res对象
|
||||
"""
|
||||
res = None
|
||||
try:
|
||||
logging.info(f"""
|
||||
【请求地址】Url: {url}
|
||||
【请求类型】Method: {method}
|
||||
【请求头】Headers: {headers}
|
||||
【请求数据】Data: {data}
|
||||
""")
|
||||
if data_type == 'params':
|
||||
res = self.session.request(method=method, url=url, params=data, headers=headers)
|
||||
elif data_type == 'data':
|
||||
res = self.session.request(method=method, url=url, data=data, headers=headers)
|
||||
elif data_type == 'json':
|
||||
res = self.session.request(method=method, url=url, json=data, headers=headers)
|
||||
elif data_type == 'file':
|
||||
res = self.session.request(method=method, url=url, files=data, headers=headers)
|
||||
else:
|
||||
logging.error('parametric_key为params、json、data、file, 不支持其他类型')
|
||||
except Exception as e:
|
||||
logging.error(f"发送请求异常:{e}")
|
||||
return res
|
||||
@@ -1,76 +0,0 @@
|
||||
# -*- coding: utf-8 -*-#
|
||||
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from pathlib import Path
|
||||
from functools import lru_cache
|
||||
|
||||
from tools.random_tool import ContextPack
|
||||
|
||||
class Settings(BaseSettings):
|
||||
model_config = SettingsConfigDict(
|
||||
env_file='.env',
|
||||
env_file_encoding='utf-8',
|
||||
extra='ignore'
|
||||
)
|
||||
|
||||
BANNER: str = """
|
||||
__ _ _ _____ ___ _____ ____ __ _____
|
||||
/ /\ | | | | | / / \ | | | |_ ( (` | |
|
||||
/_/--\ \_\_/ |_| \_\_/ |_| |_|__ _)_) |_|
|
||||
"""
|
||||
# ===================公共配置======================
|
||||
# 项目的根目录
|
||||
BASE_PATH: Path = Path(__file__).parent.parent
|
||||
# 用例所需测试文件路径
|
||||
CASE_PATH: Path = BASE_PATH / "case"
|
||||
# log路径
|
||||
LOG_PATH: Path = BASE_PATH / "logs" / (ContextPack().get_day_time + '.log')
|
||||
# 项目测试报告
|
||||
REPORT_PATH: Path = BASE_PATH / "reports"
|
||||
# 项目测试报告路径
|
||||
HTML_REPORT_PATH: Path = REPORT_PATH / (ContextPack().get_day_time + '.html')
|
||||
# api数据路径
|
||||
API_EXCEL_FILE: Path = CASE_PATH / "api" / "case.xlsx"
|
||||
API_YAML_PATH: Path = CASE_PATH / "api" / 'case.yaml'
|
||||
# UI截图路径
|
||||
UI_IMG_PATH: Path = REPORT_PATH / "img"
|
||||
|
||||
# ===================环境配置======================
|
||||
TEST_TYPE: str
|
||||
ENVIRONMENT: str
|
||||
ENVIRONMENT_HOST: str
|
||||
ENVIRONMENT_TOKEN: str
|
||||
|
||||
# ===================测试人员======================
|
||||
TESTER: str
|
||||
|
||||
# ===================公共配置======================
|
||||
DELETE_ON_OFF: bool
|
||||
EMAIL_ON_OFF: bool
|
||||
DINGDING_ON_OFF: bool
|
||||
JENKINS_URL: str
|
||||
REPORT_URL: str
|
||||
|
||||
# ===================邮件配置======================
|
||||
EMAIL_SMTP_SERVER: str
|
||||
EMAIL_SENDER: str
|
||||
EMAIL_PASSWORD: str
|
||||
EMAIL_RECEIVER: str
|
||||
|
||||
# ===================钉钉配置======================
|
||||
DINGDING_SECRET: str
|
||||
DINGDING_WEBHOOK: str
|
||||
DINGDING_AT_MOBILES: list[str]
|
||||
|
||||
# ===================浏览器配置======================
|
||||
BROWSER_TYPE: str
|
||||
|
||||
# ===================企业微信配置======================
|
||||
QIWEI_ON_OFF: bool
|
||||
WEBHOOK_URL: str
|
||||
|
||||
@lru_cache()
|
||||
def get_settings():
|
||||
return Settings()
|
||||
|
||||
settings = get_settings()
|
||||
@@ -1,34 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 必须在首次 import app 之前设置，以便 Settings / database 引擎使用 SQLite 与内存 Redis
|
||||
_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
sys.path.insert(0, _ROOT)
|
||||
|
||||
os.environ.setdefault("ENVIRONMENT", "dev")
|
||||
os.environ["TESTING"] = "1"
|
||||
os.environ["DATABASE_TYPE"] = "sqlite"
|
||||
os.environ["DATABASE_NAME"] = os.path.join(_ROOT, "pytest_fastapiadmin")
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from main import create_app
|
||||
|
||||
# 创建测试客户端（依赖上述环境变量）
|
||||
app = create_app()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_client():
|
||||
"""
|
||||
会话级 HTTP 测试客户端（与 ``conftest`` 中单例 ``app`` 对齐，只启动一次生命周期）。
|
||||
|
||||
若使用 ``module`` 作用域，多个测试文件各开一个 ``TestClient`` 会重复驱动同一
|
||||
``app`` 的 lifespan，易导致第二次 ``__enter__`` 时出现 ``CancelledError``。
|
||||
|
||||
返回:
|
||||
- TestClient: 供用例发起的同步测试客户端（yield 注入）。
|
||||
"""
|
||||
with TestClient(app) as client:
|
||||
yield client
|
||||
@@ -1,9 +0,0 @@
|
||||
locustfile = locust_files/my_locust_file.py
|
||||
headless = true
|
||||
master = true
|
||||
expect-workers = 5
|
||||
host = https://target-system
|
||||
users = 100
|
||||
spawn-rate = 10
|
||||
run-time = 10m
|
||||
tags = [Critical, Normal]
|
||||
@@ -1,23 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import pytest
|
||||
from config.setting import settings
|
||||
|
||||
# 获取Jenkins选项参数(切记参数名称不可以使用中文)
|
||||
# import os
|
||||
# TEST_TYPE = os.environ['TESTTYPE']
|
||||
# ENVIRONMENT: str = os.environ['ENVIRONMENT']
|
||||
# API_HOST: str = os.environ['APIHOST']
|
||||
# TESTER: str = os.environ['TESTER']
|
||||
|
||||
args = [
|
||||
'-sv',
|
||||
f'--html={settings.HTML_REPORT_PATH}',
|
||||
'--self-contained-html', # 静态html, 不需要额外的资源文件
|
||||
f'--log-file={settings.LOG_PATH}',
|
||||
'-W', 'ignore:Module already imported:pytest.PytestWarning'
|
||||
]
|
||||
|
||||
if __name__ == '__main__':
|
||||
# 执行测试
|
||||
pytest.main(args)
|
||||
@@ -1,48 +0,0 @@
|
||||
[pytest]
|
||||
; addopts = -v -s -q --report=_report.html --tester=Phoenixy --title=测试报告 --desc=报告描述信息 --template=2
|
||||
; addopts = -v -s -q --html=./reports/result.html --self-contained-html
|
||||
|
||||
generate_report_on_test = True
|
||||
environment_table_redact_list = ^foo$
|
||||
.*redact.*
|
||||
bar
|
||||
render_collapsed = failed,error,skipped
|
||||
|
||||
testpaths = test_case/ui
|
||||
python_files = test_*.py *_test .py test*.py
|
||||
python_classes = Test* Test_* test* test_*
|
||||
python_functions = test_* test*
|
||||
|
||||
log_file_level = INFO
|
||||
log_file_format = %(asctime)s | %(levelname)s | %(filename)s - %(funcName)s[line:%(lineno)d]: %(message)s
|
||||
log_file_date_format = %Y-%m-%d %H:%M:%S
|
||||
|
||||
log_cli = True
|
||||
log_cli_level = INFO
|
||||
; log_cli_format = %(asctime)s | %(levelname)s | %(filename)s - %(funcName)s[line:%(lineno)d]: %(message)s
|
||||
; log_cli_date_format=%Y-%m-%d %H:%M:%S
|
||||
|
||||
# --html: 生成html的测试报告,后面 需要跟上所创建的文件位置及文件名称（前提安装插件：pytest-html）
|
||||
; pytest -vs --html ./reports/result.html
|
||||
; # -n： 多线程运行（前提安装插件：pytest-xdist）
|
||||
; pytest -vs -n=2
|
||||
; # --reruns num: 失败重跑（前提安装插件：pytest-rerunfailres）
|
||||
; pytest -vs --reruns=2
|
||||
; # -x: 出现一个用例失败则停止测试
|
||||
; pytest -vs -x
|
||||
; # --maxfail: 出现几个失败才终止
|
||||
; pytest -vs --maxfail=2
|
||||
; # -k: 运行测试用例名称中包含某个字符串的测试用例，我们可以采用or表示或者，采用and表示都
|
||||
; pytest -vs -k "qiuluo"
|
||||
; pytest -vs -k "qiuluo or weiliang"
|
||||
; pytest -vs -k "qiuluo and weiliang"
|
||||
; # -m：冒烟用例执行，后面需要跟一个冒烟名称，执行user_manage这个分组
|
||||
; pytest -vs -m user_manage
|
||||
|
||||
; 下面是解决中文乱码问题：是为了在测试用例收集完成后对测试项的 name 和 nodeid 进行处理，以确保它们在控制台上的显示是正确的。
|
||||
disable_test_id_escaping_and_forfeit_all_rights_to_community_support = True
|
||||
|
||||
filterwarnings =
|
||||
error
|
||||
ignore::UserWarning
|
||||
ignore:function ham\(\) is deprecated:DeprecationWarning
|
||||
Binary file not shown.
@@ -1,119 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import datetime
|
||||
import time
|
||||
import os
|
||||
import shutil
|
||||
import pytest
|
||||
from _pytest.terminal import TerminalReporter
|
||||
import pytest_html
|
||||
import pytest_html.extras
|
||||
from pytest_metadata.plugin import metadata_key
|
||||
import logging
|
||||
|
||||
from config.setting import settings
|
||||
from tools.email_tool import EmailPack
|
||||
from tools.dingding_tool import DingTalkPack
|
||||
from tools.qiwei_tool import QiWeiPack
|
||||
|
||||
# 常量定义
|
||||
TEST_TYPE: str = settings.TEST_TYPE
|
||||
ENVIRONMENT: str = settings.ENVIRONMENT
|
||||
TESTER: str = settings.TESTER
|
||||
|
||||
"""
|
||||
module级别的测试初始化和结果收集
|
||||
class级别的测试初始化和结果收集
|
||||
function级别的测试初始化和结果收集
|
||||
"""
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def init_session():
|
||||
"""会话级别的测试初始化和结果收集"""
|
||||
if TEST_TYPE not in ['API', 'UI']:
|
||||
raise ValueError("请输入正确的自动化测试类型['API','UI']")
|
||||
if settings.DELETE_ON_OFF == 'True':
|
||||
if os.path.exists(settings.REPORT_PATH):
|
||||
shutil.rmtree(settings.REPORT_PATH)
|
||||
logging.info("历史报告数据清理完成")
|
||||
logging.info(f"{settings.BANNER}\n"
|
||||
f"开始执行 {TEST_TYPE} 自动化测试\n"
|
||||
f"测试环境: {ENVIRONMENT}\n"
|
||||
f"测试人员: {TESTER}")
|
||||
yield
|
||||
logging.info(f"========== {TEST_TYPE} 自动化测试结束 ==========")
|
||||
|
||||
def pytest_terminal_summary(terminalreporter: TerminalReporter, exitstatus: int, config: pytest.Config):
|
||||
"""收集终端测试结果"""
|
||||
stats = terminalreporter.stats
|
||||
result = {
|
||||
'total': terminalreporter._numcollected,
|
||||
'pass_num': len(stats.get('passed', [])),
|
||||
'fail_num': len(stats.get('failed', [])),
|
||||
'skip_num': len(stats.get('skipped', [])),
|
||||
'error_num': len(stats.get('error', [])),
|
||||
'rate': f"{((len(stats.get('passed', [])) / terminalreporter._numcollected * 100) if terminalreporter._numcollected else 0):.2f}%",
|
||||
'duration': f'{round(time.time() - terminalreporter._sessionstarttime, 2)}秒',
|
||||
'reprot_url': settings.REPORT_URL,
|
||||
'jenkins_url': settings.JENKINS_URL
|
||||
}
|
||||
|
||||
logging.info(f"总用例数: {result['total']} | 通过: {result['pass_num']}| 失败: {result['fail_num']} | 跳过: {result['skip_num']} | 错误: {result['error_num']} | 成功率: {result['rate']} | 总耗时: {result['duration']}")
|
||||
|
||||
# 发送邮件
|
||||
if settings.EMAIL_ON_OFF:
|
||||
EmailPack(
|
||||
fromaddr=settings.EMAIL_SENDER,
|
||||
password=settings.EMAIL_PASSWORD,
|
||||
toaddrs=[settings.EMAIL_RECEIVER],
|
||||
server_host=settings.EMAIL_SMTP_SERVER,
|
||||
).send_default_email(
|
||||
title=TEST_TYPE,
|
||||
environment=ENVIRONMENT,
|
||||
tester=TESTER,
|
||||
**result
|
||||
)
|
||||
|
||||
# 发送钉钉消息
|
||||
if settings.DINGDING_ON_OFF:
|
||||
DingTalkPack(settings.DINGDING_WEBHOOK, settings.DINGDING_SECRET).send_dingding(
|
||||
title=TEST_TYPE,
|
||||
environment=ENVIRONMENT,
|
||||
tester=TESTER,
|
||||
**result
|
||||
)
|
||||
|
||||
# 发送企业微信消息
|
||||
if settings.QIWEI_ON_OFF:
|
||||
QiWeiPack(settings.WEBHOOK_URL).send_test_report(
|
||||
title=TEST_TYPE,
|
||||
environment=ENVIRONMENT,
|
||||
tester=TESTER,
|
||||
**result
|
||||
)
|
||||
|
||||
def pytest_html_report_title(report):
|
||||
"""设置报告标题"""
|
||||
report.title = f"{TEST_TYPE}自动化测试报告"
|
||||
|
||||
def pytest_configure(config: pytest.Config):
|
||||
"""测试配置"""
|
||||
config.stash[metadata_key].update({
|
||||
"项目名称": f"{TEST_TYPE} 自动化测试",
|
||||
"测试类型": TEST_TYPE,
|
||||
"测试环境": ENVIRONMENT,
|
||||
"测试人员": TESTER,
|
||||
"开始时间": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||||
})
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_makereport(item, call: pytest.CallInfo):
|
||||
"""处理测试报告"""
|
||||
outcome = yield
|
||||
report: pytest.TestReport = outcome.get_result()
|
||||
|
||||
if report.when == "call" and report.failed:
|
||||
extras = getattr(report, 'extras', [])
|
||||
if os.path.exists(settings.LOG_PATH):
|
||||
extras.append(pytest_html.extras.url(settings.LOG_PATH, name="查看详细日志"))
|
||||
report.extras = extras
|
||||
|
||||
@@ -1,63 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import pytest
|
||||
import logging
|
||||
from typing import Dict
|
||||
from config.setting import settings
|
||||
|
||||
from tools.excel_tool import ExcelPack
|
||||
from tools.yaml_tool import YamlPack
|
||||
from tools.manage import Manage
|
||||
|
||||
manage = Manage()
|
||||
excel = ExcelPack(file_name=settings.API_EXCEL_FILE, api_host=settings.ENVIRONMENT_HOST, token=settings.ENVIRONMENT_TOKEN)
|
||||
yaml = YamlPack(yaml_path=settings.API_YAML_PATH, api_host=settings.ENVIRONMENT_HOST, token=settings.ENVIRONMENT_TOKEN)
|
||||
|
||||
|
||||
class TestExcel:
|
||||
|
||||
@pytest.mark.parametrize('case', excel.load_test_cases(), ids=lambda x: x['name'])
|
||||
def test_excel_api(self, case: Dict) -> None:
|
||||
"""Excel数据驱动的API测试类"""
|
||||
logging.info(f"开始执行用例: {case['name']}")
|
||||
|
||||
# 检查是否需要执行
|
||||
if str(case['run']).lower() != 'yes':
|
||||
logging.info(f"跳过用例: {case['name']}")
|
||||
excel.write_to_excel(row=case['row'], result=manage.create_result(case['name'], 'skip'))
|
||||
pytest.skip()
|
||||
|
||||
# 执行测试用例
|
||||
result = manage.execute_test_case(case)
|
||||
|
||||
# 写入结果
|
||||
excel.write_to_excel(row=case['row'], result=result)
|
||||
|
||||
# 断言结果
|
||||
assert result['result'] == 'pass', (
|
||||
f"用例执行失败: {case['name']}\n"
|
||||
f"请求信息: {case['request']}\n"
|
||||
f"期望结果: {case['expected']}\n"
|
||||
f"实际结果: {result['response']}"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize('case', yaml.load_test_cases(), ids=lambda x: x['name'])
|
||||
def test_yaml_api(self, case: Dict) -> None:
|
||||
"""Yaml数据驱动的API测试类"""
|
||||
logging.info(f"开始执行用例: {case['name']}")
|
||||
|
||||
# 检查是否需要执行
|
||||
if not case['run']:
|
||||
logging.info(f"跳过用例: {case['name']}")
|
||||
pytest.skip()
|
||||
|
||||
# 执行测试用例
|
||||
result = manage.execute_test_case(case)
|
||||
|
||||
# 断言结果
|
||||
assert result['result'] == 'pass', (
|
||||
f"用例执行失败: {case['name']}\n"
|
||||
f"请求信息: {case['request']}\n"
|
||||
f"期望结果: {case['expected']}\n"
|
||||
f"实际结果: {result['response']}"
|
||||
)
|
||||
@@ -1,187 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import base64
|
||||
import time
|
||||
import logging
|
||||
import datetime
|
||||
import shutil
|
||||
import pytest
|
||||
from pytest_metadata.plugin import metadata_key
|
||||
from _pytest.terminal import TerminalReporter
|
||||
|
||||
from config.setting import settings
|
||||
from tools.email_tool import EmailPack
|
||||
from tools.dingding_tool import DingTalkPack
|
||||
from tools.qiwei_tool import QiWeiPack
|
||||
|
||||
# 常量定义
|
||||
TEST_TYPE: str = settings.TEST_TYPE
|
||||
ENVIRONMENT: str = settings.ENVIRONMENT
|
||||
TESTER: str = settings.TESTER
|
||||
|
||||
# 状态常量
|
||||
success = "✓"
|
||||
fail = "✗"
|
||||
|
||||
# 全局page对象，用于在整个测试会话中共享
|
||||
GLOBAL_PAGE = None
|
||||
|
||||
"""
|
||||
module级别的测试初始化和结果收集
|
||||
class级别的测试初始化和结果收集
|
||||
function级别的测试初始化和结果收集
|
||||
"""
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def init_session():
|
||||
"""会话级别的测试初始化和结果收集 - 使用Playwright，确保唯一驱动实例"""
|
||||
if TEST_TYPE not in ['API', 'UI']:
|
||||
raise ValueError("请输入正确的自动化测试类型['API','UI']")
|
||||
if settings.DELETE_ON_OFF == 'True':
|
||||
if settings.REPORT_PATH.exists():
|
||||
shutil.rmtree(settings.REPORT_PATH)
|
||||
logging.info("历史报告数据清理完成")
|
||||
logging.info(
|
||||
f"{settings.BANNER}\n"
|
||||
f"开始执行 {TEST_TYPE} 自动化测试\n"
|
||||
f"测试环境: {ENVIRONMENT}\n"
|
||||
f"测试人员: {TESTER}"
|
||||
)
|
||||
from playwright.sync_api import sync_playwright
|
||||
playwright = sync_playwright().start()
|
||||
browser = playwright.chromium.launch(headless=False)
|
||||
context = browser.new_context(record_video_dir="./report/videos/", viewport={"width": 1920, "height": 1080}) # 生成适配视频
|
||||
page = context.new_page()
|
||||
# 更新全局page对象
|
||||
global GLOBAL_PAGE
|
||||
GLOBAL_PAGE = page
|
||||
logging.info(f"{success} ==> Playwright 浏览器实例初始化完成")
|
||||
# 返回page对象供测试用例使用
|
||||
yield page
|
||||
# 清理资源
|
||||
try:
|
||||
context.close()
|
||||
browser.close()
|
||||
playwright.stop()
|
||||
logging.info(f"{success} ==> Playwright 资源已全部释放")
|
||||
except Exception as e:
|
||||
logging.error(f"{fail} ==> Playwright 资源释放失败: {str(e)}")
|
||||
else:
|
||||
logging.info(f"========== {TEST_TYPE} 自动化测试结束 ==========")
|
||||
|
||||
def pytest_terminal_summary(terminalreporter: TerminalReporter, exitstatus: int, config: pytest.Config):
|
||||
"""收集终端测试结果"""
|
||||
stats = terminalreporter.stats
|
||||
result = {
|
||||
'total': terminalreporter._numcollected,
|
||||
'passed': len(stats.get('passed', [])),
|
||||
'failed': len(stats.get('failed', [])),
|
||||
'skipped': len(stats.get('skipped', [])),
|
||||
'error': len(stats.get('error', [])),
|
||||
'success_rate': f"{((len(stats.get('passed', [])) / terminalreporter._numcollected * 100) if terminalreporter._numcollected else 0):.2f}%",
|
||||
'duration': f'{round(time.time() - terminalreporter._sessionstarttime, 2)}秒',
|
||||
'reprot_url': settings.REPORT_URL,
|
||||
'jenkins_url': settings.JENKINS_URL
|
||||
}
|
||||
|
||||
logging.info(f"总用例数: {result['total']} | 通过: {result['passed']}| 失败: {result['failed']} | 跳过: {result['skipped']} | 错误: {result['error']} | 成功率: {result['success_rate']} | 总耗时: {result['duration']}")
|
||||
|
||||
# 发送邮件
|
||||
if settings.EMAIL_ON_OFF:
|
||||
EmailPack(
|
||||
fromaddr=settings.EMAIL_SENDER,
|
||||
password=settings.EMAIL_PASSWORD,
|
||||
toaddrs=[settings.EMAIL_RECEIVER],
|
||||
server_host=settings.EMAIL_SMTP_SERVER
|
||||
).send_default_email(
|
||||
title=TEST_TYPE,
|
||||
environment=ENVIRONMENT,
|
||||
tester=TESTER,
|
||||
**result
|
||||
)
|
||||
|
||||
# 发送钉钉消息
|
||||
if settings.DINGDING_ON_OFF:
|
||||
DingTalkPack(settings.DINGDING_WEBHOOK, settings.DINGDING_SECRET).send_dingding(
|
||||
title=TEST_TYPE,
|
||||
environment=ENVIRONMENT,
|
||||
tester=TESTER,
|
||||
**result
|
||||
)
|
||||
|
||||
# 发送企业微信消息
|
||||
if settings.QIWEI_ON_OFF:
|
||||
# 调整参数名以匹配企业微信工具的要求
|
||||
qiwei_result = {
|
||||
'total': result['total'],
|
||||
'pass_num': result['passed'],
|
||||
'fail_num': result['failed'],
|
||||
'error_num': result['error'],
|
||||
'skip_num': result['skipped'],
|
||||
'rate': result['success_rate'],
|
||||
'duration': result['duration'],
|
||||
'report_url': result['reprot_url'], # 修正拼写错误
|
||||
'jenkins_url': result['jenkins_url']
|
||||
}
|
||||
QiWeiPack(settings.WEBHOOK_URL).send_test_report(
|
||||
title=TEST_TYPE,
|
||||
environment=ENVIRONMENT,
|
||||
tester=TESTER,
|
||||
**qiwei_result
|
||||
)
|
||||
|
||||
def pytest_html_report_title(report):
|
||||
"""设置报告标题"""
|
||||
report.title = f"{TEST_TYPE}自动化测试报告"
|
||||
|
||||
def pytest_configure(config: pytest.Config):
|
||||
"""测试配置"""
|
||||
config.stash[metadata_key].update({
|
||||
"项目名称": f"{TEST_TYPE} 自动化测试",
|
||||
"测试类型": TEST_TYPE,
|
||||
"测试环境": ENVIRONMENT,
|
||||
"测试人员": TESTER,
|
||||
"开始时间": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||||
})
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo):
|
||||
pytest_html = item.config.pluginmanager.getplugin('html')
|
||||
outcome = yield
|
||||
report: pytest.TestReport = outcome.get_result()
|
||||
extras = getattr(report, 'extras', [])
|
||||
extras.append(pytest_html.extras.url(str(settings.LOG_PATH), name="查看运行日志"))
|
||||
if report.when == 'call' or report.when == "setup":
|
||||
xfail = hasattr(report, 'wasxfail')
|
||||
if (report.skipped and xfail) or (report.failed and not xfail):
|
||||
file_name = report.nodeid.replace("::", "_")+".png"
|
||||
# 使用全局page对象
|
||||
base64_img = screenshot_base64()
|
||||
if file_name:
|
||||
html = """
|
||||
<div>
|
||||
<img src="data:image/png;base64,%s" alt="screenshot" style="width:460px;height:230px;" align="right" onclick="window.open(this.src)"/>
|
||||
</div>
|
||||
""" % base64_img
|
||||
extras.append(pytest_html.extras.html(html))
|
||||
report.extras = extras
|
||||
report.description = str(item.function.__doc__)
|
||||
|
||||
def screenshot_base64():
|
||||
"""获取当前页面的截图base64编码，使用全局page对象"""
|
||||
# 生成简洁的截图文件名（年月日时分秒格式）
|
||||
settings.UI_IMG_PATH.mkdir(parents=True, exist_ok=True)
|
||||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
screenshot_path = settings.UI_IMG_PATH / f"{timestamp}.png"
|
||||
# 使用全局page对象
|
||||
global GLOBAL_PAGE
|
||||
page = GLOBAL_PAGE
|
||||
# 检查page对象是否存在
|
||||
if page and hasattr(page, 'screenshot'):
|
||||
try:
|
||||
screenshot_bytes = page.screenshot(path=screenshot_path)
|
||||
logging.info(f"{success} ==> 保存失败截图: {screenshot_path}")
|
||||
return base64.b64encode(screenshot_bytes).decode('utf-8')
|
||||
except Exception as e:
|
||||
logging.error(f"{fail} ==> 截图失败: {str(e)}")
|
||||
return ''
|
||||
|
||||
@@ -1,25 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import pytest
|
||||
from playwright.sync_api import Page
|
||||
from case.ui.pages.base_page import BasePage
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('init_session')
|
||||
class TestUI:
|
||||
|
||||
# 定义测试数据
|
||||
TEST_DATA = [{参数化1},{参数化2}]
|
||||
|
||||
@pytest.mark.ui
|
||||
@pytest.mark.parametrize("test_data", TEST_DATA)
|
||||
def test_login_scenarios(self, init_session: Page, test_data: Dict[str, Any]):
|
||||
"""参数化测试不同的登录场景"""
|
||||
emposat_page = BasePage(init_session)
|
||||
|
||||
login_result = emposat_page.login(test_data=test_data)
|
||||
assert login_result == True, "登录失败"
|
||||
|
||||
logout_result = emposat_page.logout()
|
||||
assert logout_result == True, "退出登录失败"
|
||||
@@ -1,143 +0,0 @@
|
||||
"""
|
||||
类型转换测试。
|
||||
|
||||
测试MySQL、PostgreSQL、Python和前端TypeScript的代码生成字段类型转换正确性。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from app.common.constant import GenConstant
|
||||
from app.config.setting import settings
|
||||
from app.plugin.module_generator.gencode.tools.gen_util import GenUtils
|
||||
from app.plugin.module_generator.gencode.tools.jinja2_template_util import Jinja2TemplateUtil
|
||||
|
||||
|
||||
def test_mysql_type_conversion():
|
||||
"""
|
||||
测试MySQL类型转换
|
||||
"""
|
||||
# 保存原始数据库类型
|
||||
original_db_type = settings.DATABASE_TYPE
|
||||
try:
|
||||
# 设置为MySQL
|
||||
settings.DATABASE_TYPE = "mysql"
|
||||
|
||||
# 测试MySQL类型
|
||||
test_cases = [
|
||||
# (数据库类型, 期望的Python类型, 期望的TypeScript类型)
|
||||
("INT", "int", "number"),
|
||||
("BIGINT", "int", "number"),
|
||||
("TINYINT", "int", "number"),
|
||||
("TINYINT(1)", "bool", "boolean"),
|
||||
("FLOAT", "float", "number"),
|
||||
("DOUBLE", "float", "number"),
|
||||
("DECIMAL(10,2)", "Decimal", "string"),
|
||||
("CHAR(10)", "str", "string"),
|
||||
("VARCHAR(255)", "str", "string"),
|
||||
("TEXT", "str", "string"),
|
||||
("DATE", "date", "string"),
|
||||
("TIME", "time", "string"),
|
||||
("DATETIME", "datetime", "string"),
|
||||
("TIMESTAMP", "datetime", "string"),
|
||||
("YEAR", "int", "number"),
|
||||
("JSON", "dict", "Record<string, unknown>"),
|
||||
("ENUM('a','b','c')", "str", "string"),
|
||||
]
|
||||
|
||||
for db_type, expected_python_type, expected_ts_type in test_cases:
|
||||
# 测试数据库类型提取
|
||||
extracted_type = GenUtils.get_db_type(db_type)
|
||||
# 测试Python类型映射
|
||||
python_type = GenConstant.DB_TO_PYTHON.get(extracted_type, "str")
|
||||
assert python_type == expected_python_type, f"MySQL {db_type} -> Python {python_type}, expected {expected_python_type}"
|
||||
# 测试TypeScript类型映射
|
||||
ts_type = Jinja2TemplateUtil.python_type_to_ts_type(python_type)
|
||||
assert ts_type == expected_ts_type, f"Python {python_type} -> TypeScript {ts_type}, expected {expected_ts_type}"
|
||||
finally:
|
||||
# 恢复原始数据库类型
|
||||
settings.DATABASE_TYPE = original_db_type
|
||||
|
||||
|
||||
def test_postgres_type_conversion():
|
||||
"""
|
||||
测试PostgreSQL类型转换
|
||||
"""
|
||||
# 保存原始数据库类型
|
||||
original_db_type = settings.DATABASE_TYPE
|
||||
try:
|
||||
# 设置为PostgreSQL
|
||||
settings.DATABASE_TYPE = "postgres"
|
||||
|
||||
# 测试PostgreSQL类型
|
||||
test_cases = [
|
||||
# (数据库类型, 期望的Python类型, 期望的TypeScript类型)
|
||||
("INTEGER", "int", "number"),
|
||||
("BIGINT", "int", "number"),
|
||||
("SMALLINT", "int", "number"),
|
||||
("REAL", "float", "number"),
|
||||
("DOUBLE PRECISION", "float", "number"),
|
||||
("NUMERIC(10,2)", "Decimal", "string"),
|
||||
("VARCHAR(255)", "str", "string"),
|
||||
("CHAR(10)", "str", "string"),
|
||||
("TEXT", "str", "string"),
|
||||
("DATE", "date", "string"),
|
||||
("TIME", "time", "string"),
|
||||
("TIMESTAMP", "datetime", "string"),
|
||||
("TIMESTAMP WITH TIME ZONE", "datetime", "string"),
|
||||
("BOOLEAN", "bool", "boolean"),
|
||||
("JSONB", "dict", "Record<string, unknown>"),
|
||||
("UUID", "str", "string"),
|
||||
("ARRAY[INTEGER]", "list", "unknown[]"),
|
||||
("POINT", "list", "unknown[]"),
|
||||
]
|
||||
|
||||
for db_type, expected_python_type, expected_ts_type in test_cases:
|
||||
# 测试数据库类型提取
|
||||
extracted_type = GenUtils.get_db_type(db_type)
|
||||
# 测试Python类型映射
|
||||
python_type = GenConstant.DB_TO_PYTHON.get(extracted_type, "str")
|
||||
assert python_type == expected_python_type, f"PostgreSQL {db_type} -> Python {python_type}, expected {expected_python_type}"
|
||||
# 测试TypeScript类型映射
|
||||
ts_type = Jinja2TemplateUtil.python_type_to_ts_type(python_type)
|
||||
assert ts_type == expected_ts_type, f"Python {python_type} -> TypeScript {ts_type}, expected {expected_ts_type}"
|
||||
finally:
|
||||
# 恢复原始数据库类型
|
||||
settings.DATABASE_TYPE = original_db_type
|
||||
|
||||
|
||||
def test_special_type_handling():
|
||||
"""
|
||||
测试特殊类型处理
|
||||
"""
|
||||
# 测试tinyint(1)特殊处理
|
||||
assert GenUtils.get_db_type("TINYINT(1)") == "boolean"
|
||||
|
||||
# 测试数组类型处理
|
||||
assert GenUtils.get_db_type("INTEGER[]") == "array"
|
||||
assert GenUtils.get_db_type("TEXT[]") == "array"
|
||||
|
||||
# 测试带COLLATE的类型
|
||||
assert GenUtils.get_db_type("VARCHAR(255) COLLATE utf8mb4_unicode_ci") == "varchar"
|
||||
|
||||
# 测试带UNSIGNED的类型
|
||||
assert GenUtils.get_db_type("INT UNSIGNED") == "int"
|
||||
|
||||
|
||||
def test_sqlalchemy_type_mapping():
|
||||
"""
|
||||
测试SQLAlchemy类型映射
|
||||
"""
|
||||
# 测试MySQL SQLAlchemy类型
|
||||
settings.DATABASE_TYPE = "mysql"
|
||||
assert Jinja2TemplateUtil.get_sqlalchemy_type("INT") == "Integer"
|
||||
assert Jinja2TemplateUtil.get_sqlalchemy_type("VARCHAR(255)") == "String(255)"
|
||||
assert Jinja2TemplateUtil.get_sqlalchemy_type("TINYINT(1)") == "Boolean"
|
||||
|
||||
# 测试PostgreSQL SQLAlchemy类型
|
||||
settings.DATABASE_TYPE = "postgres"
|
||||
assert Jinja2TemplateUtil.get_sqlalchemy_type("INTEGER") == "Integer"
|
||||
assert Jinja2TemplateUtil.get_sqlalchemy_type("VARCHAR(255)") == "String(255)"
|
||||
assert Jinja2TemplateUtil.get_sqlalchemy_type("BOOLEAN") == "Boolean"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main(["-v", "tests/test_type_conversion.py"])
|
||||
@@ -1,323 +0,0 @@
|
||||
"""
|
||||
全面的类型转换测试。
|
||||
|
||||
测试MySQL、PostgreSQL、Python和前端TypeScript的代码生成字段类型转换正确性，覆盖所有支持的类型。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from app.common.constant import GenConstant
|
||||
from app.config.setting import settings
|
||||
from app.plugin.module_generator.gencode.tools.gen_util import GenUtils
|
||||
from app.plugin.module_generator.gencode.tools.jinja2_template_util import Jinja2TemplateUtil
|
||||
|
||||
|
||||
def test_mysql_complete_type_conversion():
|
||||
"""
|
||||
测试MySQL完整类型转换
|
||||
"""
|
||||
# 保存原始数据库类型
|
||||
original_db_type = settings.DATABASE_TYPE
|
||||
try:
|
||||
# 设置为MySQL
|
||||
settings.DATABASE_TYPE = "mysql"
|
||||
|
||||
# 测试所有MySQL类型
|
||||
test_cases = [
|
||||
# 整数类型
|
||||
("TINYINT", "int", "number"),
|
||||
("TINYINT(1)", "bool", "boolean"),
|
||||
("TINYINT UNSIGNED", "int", "number"),
|
||||
("SMALLINT", "int", "number"),
|
||||
("SMALLINT UNSIGNED", "int", "number"),
|
||||
("MEDIUMINT", "int", "number"),
|
||||
("MEDIUMINT UNSIGNED", "int", "number"),
|
||||
("INT", "int", "number"),
|
||||
("INT UNSIGNED", "int", "number"),
|
||||
("BIGINT", "int", "number"),
|
||||
("BIGINT UNSIGNED", "int", "number"),
|
||||
|
||||
# 浮点类型
|
||||
("FLOAT", "float", "number"),
|
||||
("DOUBLE", "float", "number"),
|
||||
("DECIMAL", "Decimal", "string"),
|
||||
("DECIMAL(10,2)", "Decimal", "string"),
|
||||
("NUMERIC", "Decimal", "string"),
|
||||
("NUMERIC(10,2)", "Decimal", "string"),
|
||||
|
||||
# 字符串类型
|
||||
("CHAR", "str", "string"),
|
||||
("CHAR(10)", "str", "string"),
|
||||
("VARCHAR", "str", "string"),
|
||||
("VARCHAR(255)", "str", "string"),
|
||||
("VARCHAR(255) COLLATE utf8mb4_unicode_ci", "str", "string"),
|
||||
("TINYTEXT", "str", "string"),
|
||||
("TEXT", "str", "string"),
|
||||
("MEDIUMTEXT", "str", "string"),
|
||||
("LONGTEXT", "str", "string"),
|
||||
|
||||
# 二进制类型
|
||||
("BINARY", "bytes", "string"),
|
||||
("BINARY(10)", "bytes", "string"),
|
||||
("VARBINARY", "bytes", "string"),
|
||||
("VARBINARY(255)", "bytes", "string"),
|
||||
("TINYBLOB", "bytes", "string"),
|
||||
("BLOB", "bytes", "string"),
|
||||
("MEDIUMBLOB", "bytes", "string"),
|
||||
("LONGBLOB", "bytes", "string"),
|
||||
|
||||
# 日期时间类型
|
||||
("DATE", "date", "string"),
|
||||
("TIME", "time", "string"),
|
||||
("DATETIME", "datetime", "string"),
|
||||
("TIMESTAMP", "datetime", "string"),
|
||||
("YEAR", "int", "number"),
|
||||
|
||||
# 其他类型
|
||||
("JSON", "dict", "Record<string, unknown>"),
|
||||
("ENUM('a','b','c')", "str", "string"),
|
||||
("SET('a','b','c')", "str", "string"),
|
||||
("BIT", "int", "number"),
|
||||
("BIT(1)", "int", "number"),
|
||||
|
||||
# 空间数据类型
|
||||
("GEOMETRY", "bytes", "string"),
|
||||
("POINT", "bytes", "string"),
|
||||
("LINESTRING", "bytes", "string"),
|
||||
("POLYGON", "bytes", "string"),
|
||||
("MULTIPOINT", "bytes", "string"),
|
||||
("MULTILINESTRING", "bytes", "string"),
|
||||
("MULTIPOLYGON", "bytes", "string"),
|
||||
("GEOMETRYCOLLECTION", "bytes", "string"),
|
||||
]
|
||||
|
||||
for db_type, expected_python_type, expected_ts_type in test_cases:
|
||||
# 测试数据库类型提取
|
||||
extracted_type = GenUtils.get_db_type(db_type)
|
||||
# 测试Python类型映射
|
||||
|
||||
# 特殊处理几何类型，根据数据库类型选择不同的映射
|
||||
if extracted_type in ["point", "line", "linestring", "polygon", "multipoint", "multilinestring", "multipolygon", "geometrycollection", "geometry"]:
|
||||
python_type = "bytes" # MySQL几何类型映射为bytes
|
||||
else:
|
||||
python_type = GenConstant.DB_TO_PYTHON.get(extracted_type, "str")
|
||||
|
||||
assert python_type == expected_python_type, f"MySQL {db_type} -> Python {python_type}, expected {expected_python_type}"
|
||||
# 测试TypeScript类型映射
|
||||
ts_type = Jinja2TemplateUtil.python_type_to_ts_type(python_type)
|
||||
assert ts_type == expected_ts_type, f"Python {python_type} -> TypeScript {ts_type}, expected {expected_ts_type}"
|
||||
finally:
|
||||
# 恢复原始数据库类型
|
||||
settings.DATABASE_TYPE = original_db_type
|
||||
|
||||
|
||||
def test_postgres_complete_type_conversion():
|
||||
"""
|
||||
测试PostgreSQL完整类型转换
|
||||
"""
|
||||
# 保存原始数据库类型
|
||||
original_db_type = settings.DATABASE_TYPE
|
||||
try:
|
||||
# 设置为PostgreSQL
|
||||
settings.DATABASE_TYPE = "postgres"
|
||||
|
||||
# 测试所有PostgreSQL类型
|
||||
test_cases = [
|
||||
# 整数类型
|
||||
("SMALLINT", "int", "number"),
|
||||
("INT2", "int", "number"),
|
||||
("INTEGER", "int", "number"),
|
||||
("INT4", "int", "number"),
|
||||
("BIGINT", "int", "number"),
|
||||
("INT8", "int", "number"),
|
||||
|
||||
# 浮点类型
|
||||
("REAL", "float", "number"),
|
||||
("DOUBLE PRECISION", "float", "number"),
|
||||
("FLOAT8", "float", "number"),
|
||||
("NUMERIC", "Decimal", "string"),
|
||||
("NUMERIC(10,2)", "Decimal", "string"),
|
||||
("DECIMAL", "Decimal", "string"),
|
||||
("DECIMAL(10,2)", "Decimal", "string"),
|
||||
|
||||
# 字符串类型
|
||||
("CHAR", "str", "string"),
|
||||
("CHAR(10)", "str", "string"),
|
||||
("VARCHAR", "str", "string"),
|
||||
("VARCHAR(255)", "str", "string"),
|
||||
("CHARACTER", "str", "string"),
|
||||
("CHARACTER(10)", "str", "string"),
|
||||
("CHARACTER VARYING", "str", "string"),
|
||||
("CHARACTER VARYING(255)", "str", "string"),
|
||||
("TEXT", "str", "string"),
|
||||
("CITEXT", "str", "string"),
|
||||
|
||||
# 二进制类型
|
||||
("BYTEA", "bytes", "string"),
|
||||
|
||||
# 日期时间类型
|
||||
("DATE", "date", "string"),
|
||||
("TIME", "time", "string"),
|
||||
("TIME WITHOUT TIME ZONE", "time", "string"),
|
||||
("TIME WITH TIME ZONE", "time", "string"),
|
||||
("TIMETZ", "time", "string"),
|
||||
("TIMESTAMP", "datetime", "string"),
|
||||
("TIMESTAMP WITHOUT TIME ZONE", "datetime", "string"),
|
||||
("TIMESTAMP WITH TIME ZONE", "datetime", "string"),
|
||||
("TIMESTAMPTZ", "datetime", "string"),
|
||||
("INTERVAL", "timedelta", "string"),
|
||||
|
||||
# 布尔类型
|
||||
("BOOLEAN", "bool", "boolean"),
|
||||
("BOOL", "bool", "boolean"),
|
||||
|
||||
# JSON类型
|
||||
("JSON", "dict", "Record<string, unknown>"),
|
||||
("JSONB", "dict", "Record<string, unknown>"),
|
||||
|
||||
# 数组类型
|
||||
("INTEGER[]", "list", "unknown[]"),
|
||||
("ARRAY[INTEGER]", "list", "unknown[]"),
|
||||
("TEXT[]", "list", "unknown[]"),
|
||||
("ARRAY[TEXT]", "list", "unknown[]"),
|
||||
|
||||
# 几何类型
|
||||
("POINT", "list", "unknown[]"),
|
||||
("LINE", "list", "unknown[]"),
|
||||
("LSEG", "list", "unknown[]"),
|
||||
("BOX", "list", "unknown[]"),
|
||||
("PATH", "list", "unknown[]"),
|
||||
("POLYGON", "list", "unknown[]"),
|
||||
("CIRCLE", "list", "unknown[]"),
|
||||
|
||||
# 其他类型
|
||||
("UUID", "str", "string"),
|
||||
("INET", "str", "string"),
|
||||
("CIDR", "str", "string"),
|
||||
("MACADDR", "str", "string"),
|
||||
("BIT", "int", "number"),
|
||||
("BIT VARYING", "int", "number"),
|
||||
("VARBIT", "int", "number"),
|
||||
("TSVECTOR", "str", "string"),
|
||||
("TSQUERY", "str", "string"),
|
||||
("XML", "str", "string"),
|
||||
("MONEY", "Decimal", "string"),
|
||||
("OID", "int", "number"),
|
||||
("REGPROC", "str", "string"),
|
||||
("REGCLASS", "str", "string"),
|
||||
("REGTYPE", "str", "string"),
|
||||
("REGROLE", "str", "string"),
|
||||
("REGNAMESPACE", "str", "string"),
|
||||
("INT2VECTOR", "list", "unknown[]"),
|
||||
("OIDVECTOR", "list", "unknown[]"),
|
||||
("PG_LSN", "int", "number"),
|
||||
("TXID_SNAPSHOT", "str", "string"),
|
||||
("PG_NODE_TREE", "str", "string"),
|
||||
|
||||
# 范围类型
|
||||
("INT4RANGE", "list", "unknown[]"),
|
||||
("INT8RANGE", "list", "unknown[]"),
|
||||
("TSRANGE", "list", "unknown[]"),
|
||||
("TSTZRANGE", "list", "unknown[]"),
|
||||
("DATERANGE", "list", "unknown[]"),
|
||||
]
|
||||
|
||||
for db_type, expected_python_type, expected_ts_type in test_cases:
|
||||
# 测试数据库类型提取
|
||||
extracted_type = GenUtils.get_db_type(db_type)
|
||||
# 测试Python类型映射
|
||||
|
||||
# 特殊处理几何类型，根据数据库类型选择不同的映射
|
||||
if extracted_type in ["point", "line", "lseg", "box", "path", "polygon", "circle"]:
|
||||
python_type = "list" # PostgreSQL几何类型映射为list
|
||||
else:
|
||||
python_type = GenConstant.DB_TO_PYTHON.get(extracted_type, "str")
|
||||
|
||||
assert python_type == expected_python_type, f"PostgreSQL {db_type} -> Python {python_type}, expected {expected_python_type}"
|
||||
# 测试TypeScript类型映射
|
||||
ts_type = Jinja2TemplateUtil.python_type_to_ts_type(python_type)
|
||||
assert ts_type == expected_ts_type, f"Python {python_type} -> TypeScript {ts_type}, expected {expected_ts_type}"
|
||||
finally:
|
||||
# 恢复原始数据库类型
|
||||
settings.DATABASE_TYPE = original_db_type
|
||||
|
||||
|
||||
def test_sqlalchemy_type_comprehensive():
|
||||
"""
|
||||
测试SQLAlchemy类型映射的完整性
|
||||
"""
|
||||
# 测试MySQL SQLAlchemy类型
|
||||
settings.DATABASE_TYPE = "mysql"
|
||||
mysql_test_cases = [
|
||||
("INT", "Integer"),
|
||||
("BIGINT", "BigInteger"),
|
||||
("TINYINT", "SmallInteger"),
|
||||
("TINYINT(1)", "Boolean"),
|
||||
("FLOAT", "Float"),
|
||||
("DOUBLE", "Float"),
|
||||
("DECIMAL", "DECIMAL"),
|
||||
("DECIMAL(10,2)", "DECIMAL(10,2)"),
|
||||
("CHAR", "CHAR(255)"),
|
||||
("CHAR(10)", "CHAR(10)"),
|
||||
("VARCHAR", "String(255)"),
|
||||
("VARCHAR(255)", "String(255)"),
|
||||
("TEXT", "Text"),
|
||||
("DATE", "Date"),
|
||||
("TIME", "Time"),
|
||||
("DATETIME", "DateTime"),
|
||||
("TIMESTAMP", "TIMESTAMP"),
|
||||
("YEAR", "Integer"),
|
||||
("JSON", "JSON"),
|
||||
("ENUM('a','b','c')", "Enum"),
|
||||
]
|
||||
|
||||
for db_type, expected_sqlalchemy_type in mysql_test_cases:
|
||||
sqlalchemy_type = Jinja2TemplateUtil.get_sqlalchemy_type(db_type)
|
||||
assert sqlalchemy_type == expected_sqlalchemy_type, f"MySQL {db_type} -> SQLAlchemy {sqlalchemy_type}, expected {expected_sqlalchemy_type}"
|
||||
|
||||
# 测试PostgreSQL SQLAlchemy类型
|
||||
settings.DATABASE_TYPE = "postgres"
|
||||
postgres_test_cases = [
|
||||
("INTEGER", "Integer"),
|
||||
("BIGINT", "BigInteger"),
|
||||
("SMALLINT", "SmallInteger"),
|
||||
("REAL", "Float"),
|
||||
("DOUBLE PRECISION", "Float"),
|
||||
("NUMERIC", "Numeric"),
|
||||
("NUMERIC(10,2)", "Numeric(10,2)"),
|
||||
("VARCHAR", "String(255)"),
|
||||
("VARCHAR(255)", "String(255)"),
|
||||
("TEXT", "Text"),
|
||||
("DATE", "Date"),
|
||||
("TIME", "Time"),
|
||||
("TIMESTAMP", "DateTime"),
|
||||
("BOOLEAN", "Boolean"),
|
||||
("JSONB", "JSONB"),
|
||||
("UUID", "Uuid"),
|
||||
("BYTEA", "LargeBinary"),
|
||||
]
|
||||
|
||||
for db_type, expected_sqlalchemy_type in postgres_test_cases:
|
||||
sqlalchemy_type = Jinja2TemplateUtil.get_sqlalchemy_type(db_type)
|
||||
assert sqlalchemy_type == expected_sqlalchemy_type, f"PostgreSQL {db_type} -> SQLAlchemy {sqlalchemy_type}, expected {expected_sqlalchemy_type}"
|
||||
|
||||
|
||||
def test_edge_cases():
|
||||
"""
|
||||
测试边缘情况
|
||||
"""
|
||||
# 测试空类型
|
||||
assert Jinja2TemplateUtil.python_type_to_ts_type(None) == "string"
|
||||
assert Jinja2TemplateUtil.python_type_to_ts_type("") == "string"
|
||||
assert Jinja2TemplateUtil.python_type_to_ts_type(" ") == "string"
|
||||
|
||||
# 测试未知类型
|
||||
assert Jinja2TemplateUtil.python_type_to_ts_type("unknown_type") == "string"
|
||||
|
||||
# 测试大小写不敏感
|
||||
assert GenUtils.get_db_type("VARCHAR(255)") == "varchar"
|
||||
assert GenUtils.get_db_type("tinyint(1)") == "boolean"
|
||||
assert GenUtils.get_db_type("array[integer]") == "array"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main(["-v", "tests/test_type_conversion_comprehensive.py"])
|
||||
@@ -1,101 +0,0 @@
|
||||
# -*- 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
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -1,89 +0,0 @@
|
||||
# -*- 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()
|
||||
|
||||
@@ -1,118 +0,0 @@
|
||||
# -*- 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)
|
||||
@@ -1,56 +0,0 @@
|
||||
# -*- 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
|
||||
|
||||
|
||||
@@ -1,182 +0,0 @@
|
||||
|
||||
|
||||
# -*- 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)
|
||||
|
||||
|
||||
@@ -1,281 +0,0 @@
|
||||
# -*- 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))
|
||||
@@ -1,33 +0,0 @@
|
||||
# -*- 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
|
||||
|
||||
Reference in New Issue
Block a user