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

目录

• 前言
• 使用方法
• 代码实现
• 1. 安装依赖
• 2. 主代码
• 功能解析
• 1. 界面设计
• 2. 异步请求
• 3. 响应展示
• 运行效果
• 总结

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

我们将实现以下功能：

请求方法选择器

添加了一个下拉框 QComboBox，用户可以选择 GET、POST、PUT、DELETE 或 PATCH。

动态请求方法

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

异常处理

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

在界面上输入请求的 URL。

选择所需的 请求方法（如 GET、POST 等）。

输入 请求头 和 请求参数（JSON 格式）。

设置 并发请求次数，点击“开始测试”。

查看结果表格中每个请求的序号、状态码和响应时间。

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

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

pip install pyqt5 aiohttp

以下是完整的代码实现：

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

使用 PyQt5 构建界面，布局由 QVBoxLayout 和 QHBoxLayout 组合，模块化分为“配置区”和“结果区”。

支持输入 URL、请求头、请求参数、以及指定发送次数。

使用 aiohttp.ClientSession 实现非阻塞的 HTTP 请求。

通过 asyncio.gather 并发发送多个请求，收集结果。

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

启动工具后，用户可以在界面上输入接口参数，例如 URL、请求头、请求体等。

点击“开始测试”后，工具会并发发送指定次数的请求，并实时展示结果。

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

到此这篇关于使用python和asyncio打造接口并发测试GUI工具的文章就介绍到这了,更多相关python asyncio接口并发测试内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• python实现接口并发测试脚本
• Python+threading模块对单个接口进行并发测试
• 详解Python如何精确控制asyncio并发运行多个任务
• 使用python请求接口方式(可进行并发测试)