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

news2025/4/25 10:50:37

接口并发测试是测试工程师日常工作中的重要一环，而一个直观的 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 QFont

class 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 0
                return index, response.status, elapsed
        except Exception as e:
            return index, f"Error: {str(e)}", 0

    async 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)}"))
            return

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