使用python和asyncio打造接口并發(fā)測試GUI工具

更新時間：2025年04月25日 08:39:15 作者：Python測試之道

接口并發(fā)測試是測試工程師日常工作中的重要一環(huán),而一個直觀的 GUI 工具能有效提升工作效率和體驗,下面我們就來看看如何通過 PyQt5 和 asyncio打造一個這樣的工具吧

前言

接口并發(fā)測試是測試工程師日常工作中的重要一環(huán)，而一個直觀的 GUI 工具能有效提升工作效率和體驗。本篇文章將帶你用 PyQt5 和 asyncio 從零實現一個美觀且功能實用的接口并發(fā)測試工具。

我們將實現以下功能：

請求方法選擇器

添加了一個下拉框 QComboBox，用戶可以選擇 GET、POST、PUT、DELETE 或 PATCH。

動態(tài)請求方法

根據用戶選擇的請求方法，在 send_request 函數中動態(tài)調用對應的 aiohttp 方法（如 session.get 或 session.post）。

異常處理

如果用戶選擇了不支持的請求方法，會返回 "Unsupported Method" 錯誤。

使用方法

在界面上輸入請求的 URL。

選擇所需的請求方法（如 GET、POST 等）。

輸入請求頭和請求參數（JSON 格式）。

設置并發(fā)請求次數，點擊“開始測試”。

查看結果表格中每個請求的序號、狀態(tài)碼和響應時間。

下面是完整的代碼實現以及詳細的注釋，幫助你快速上手。

代碼實現

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("接口并發(fā)測試工具")
        self.setGeometry(100, 100, 800, 600)
        self.setFont(QFont("Arial", 10))

        # === 接口配置區(qū) ===
        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("發(fā)送次數:")
        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)

        # === 結果展示區(qū) ===
        results_label = QLabel("測試結果:")
        self.results_table = QTableWidget()
        self.results_table.setColumnCount(3)
        self.results_table.setHorizontalHeaderLabels(["請求序號", "狀態(tài)碼", "響應時間 (秒)"])
        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):
        """發(fā)送單個 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):
        """啟動并發(fā)請求"""
        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_())