当前位置：首页 > news >正文

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

news 2025/7/1 16:17:58

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

我们将实现以下功能：

请求方法选择器
添加了一个下拉框 QComboBox，用户可以选择 GET、POST、PUT、DELETE 或 PATCH。
动态请求方法
根据用户选择的请求方法，在 send_request 函数中动态调用对应的 aiohttp 方法（如 session.get 或 session.post）。
异常处理
如果用户选择了不支持的请求方法，会返回 "Unsupported Method" 错误。

使用方法

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

下面是完整的代码实现以及详细的注释，帮助你快速上手。
在这里插入图片描述

代码实现

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_())