用 PyQt5 和 asyncio 打造接口并发测试 GUI 工具

接口并发测试是测试工程师日常工作中的重要一环，而一个直观的 GUI 工具能有效提升工作效率和体验。本篇文章将带你用 PyQt5 和 asyncio 从零实现一个美观且功能实用的接口并发测试工具。

我们将实现以下功能：

1. 请求方法选择器
   添加了一个下拉框 QComboBox，用户可以选择 GET、POST、PUT、DELETE 或 PATCH。

2. 动态请求方法
   根据用户选择的请求方法，在 send_request 函数中动态调用对应的 aiohttp 方法（如 session.get 或 session.post）。

3. 异常处理
   如果用户选择了不支持的请求方法，会返回 "Unsupported Method" 错误。

---

使用方法

1. 在界面上输入请求的 URL。
2. 选择所需的 请求方法（如 GET、POST 等）。
3. 输入 请求头 和 请求参数（JSON 格式）。
4. 设置 并发请求次数，点击“开始测试”。
5. 查看结果表格中每个请求的序号、状态码和响应时间。

下面是完整的代码实现以及详细的注释，帮助你快速上手。

---

代码实现

1. 安装依赖

在开始之前，请确保安装了必要的依赖库：

pip install pyqt5 aiohttp

---

2. 主代码

以下是完整的代码实现：

import sys
import asyncio
import aiohttp
from PyQt5.QtWidgets import (QApplication, QWidget, QLabel, QLineEdit, QTextEdit, QVBoxLayout, QHBoxLayout, QPushButton, QSpinBox, QTableWidget, QTableWidgetItem
)
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QFontclass AsyncHttpTester(QWidget):def __init__(self):super().__init__()self.init_ui()def init_ui(self):"""初始化用户界面"""self.setWindowTitle("接口并发测试工具")self.setGeometry(100, 100, 800, 600)self.setFont(QFont("Arial", 10))# === 接口配置区 ===url_label = QLabel("请求 URL:")self.url_input = QLineEdit()self.url_input.setPlaceholderText("请输入接口 URL")headers_label = QLabel("请求头 (JSON 格式):")self.headers_input = QTextEdit()self.headers_input.setPlaceholderText('例如：{"Content-Type": "application/json"}')params_label = QLabel("请求参数 (JSON 格式):")self.params_input = QTextEdit()self.params_input.setPlaceholderText('例如：{"key": "value"}')times_label = QLabel("发送次数:")self.times_input = QSpinBox()self.times_input.setRange(1, 1000)self.times_input.setValue(1)# === 开始按钮 ===self.start_button = QPushButton("开始测试")self.start_button.clicked.connect(self.start_test)# === 结果展示区 ===results_label = QLabel("测试结果:")self.results_table = QTableWidget()self.results_table.setColumnCount(3)self.results_table.setHorizontalHeaderLabels(["请求序号", "状态码", "响应时间 (秒)"])self.results_table.setColumnWidth(0, 100)self.results_table.setColumnWidth(1, 100)self.results_table.setColumnWidth(2, 150)# === 布局 ===layout = QVBoxLayout()# 接口配置布局config_layout = QVBoxLayout()config_layout.addWidget(url_label)config_layout.addWidget(self.url_input)config_layout.addWidget(headers_label)config_layout.addWidget(self.headers_input)config_layout.addWidget(params_label)config_layout.addWidget(self.params_input)config_layout.addWidget(times_label)config_layout.addWidget(self.times_input)# 添加开始按钮config_layout.addWidget(self.start_button)# 结果展示布局results_layout = QVBoxLayout()results_layout.addWidget(results_label)results_layout.addWidget(self.results_table)# 整合布局layout.addLayout(config_layout)layout.addLayout(results_layout)self.setLayout(layout)async def send_request(self, session, url, headers, params, index):"""发送单个 HTTP 请求"""try:async with session.post(url, json=params, headers=headers) as response:elapsed = response.elapsed.total_seconds() if response.elapsed else 0return index, response.status, elapsedexcept Exception as e:return index, f"Error: {str(e)}", 0async def start_async_requests(self, url, headers, params, times):"""启动并发请求"""tasks = []async with aiohttp.ClientSession() as session:for i in range(times):tasks.append(self.send_request(session, url, headers, params, i + 1))return await asyncio.gather(*tasks)def start_test(self):"""开始测试按钮事件"""url = self.url_input.text().strip()try:headers = eval(self.headers_input.toPlainText().strip()) if self.headers_input.toPlainText().strip() else {}params = eval(self.params_input.toPlainText().strip()) if self.params_input.toPlainText().strip() else {}except Exception as e:self.results_table.setRowCount(0)self.results_table.setRowCount(1)self.results_table.setItem(0, 0, QTableWidgetItem("Error"))self.results_table.setItem(0, 1, QTableWidgetItem(f"Invalid headers/params: {str(e)}"))returntimes = self.times_input.value()if not url:self.results_table.setRowCount(0)self.results_table.setRowCount(1)self.results_table.setItem(0, 0, QTableWidgetItem("Error"))self.results_table.setItem(0, 1, QTableWidgetItem("URL 不能为空"))return# 清空结果表self.results_table.setRowCount(0)# 启动异步任务loop = asyncio.get_event_loop()results = loop.run_until_complete(self.start_async_requests(url, headers, params, times))# 更新结果表self.results_table.setRowCount(len(results))for i, (index, status, elapsed) in enumerate(results):self.results_table.setItem(i, 0, QTableWidgetItem(str(index)))self.results_table.setItem(i, 1, QTableWidgetItem(str(status)))self.results_table.setItem(i, 2, QTableWidgetItem(f"{elapsed:.2f}"))if __name__ == "__main__":app = QApplication(sys.argv)tester = AsyncHttpTester()tester.show()sys.exit(app.exec_())

---

功能解析

1. 界面设计

• 使用 PyQt5 构建界面，布局由 QVBoxLayout 和 QHBoxLayout 组合，模块化分为“配置区”和“结果区”。
• 支持输入 URL、请求头、请求参数、以及指定发送次数。

2. 异步请求

• 使用 aiohttp.ClientSession 实现非阻塞的 HTTP 请求。
• 通过 asyncio.gather 并发发送多个请求，收集结果。

3. 响应展示

• 结果以表格形式展示，包含请求序号、状态码、响应时间，方便对比和分析。

---

运行效果

1. 启动工具后，用户可以在界面上输入接口参数，例如 URL、请求头、请求体等。
2. 点击“开始测试”后，工具会并发发送指定次数的请求，并实时展示结果。

---

总结

通过 PyQt5 和 asyncio，我们成功实现了一个美观实用的接口并发测试工具。在这个项目中，测试工程师可以直观地配置接口参数并分析响应结果，同时也能深入理解 Python 的异步编程原理。

赶紧试试吧！