接口并發測試是測試工程師日常工作中的重要一環，而一個直觀的 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_())
功能解析
1. 界面設計
- 使用 PyQt5 構建界面，布局由
QVBoxLayout和QHBoxLayout組合，模塊化分為“配置區”和“結果區”。 - 支持輸入 URL、請求頭、請求參數、以及指定發送次數。
2. 異步請求
- 使用
aiohttp.ClientSession實現非阻塞的 HTTP 請求。 - 通過
asyncio.gather并發發送多個請求，收集結果。
3. 響應展示
- 結果以表格形式展示，包含請求序號、狀態碼、響應時間，方便對比和分析。
運行效果
- 啟動工具后，用戶可以在界面上輸入接口參數，例如 URL、請求頭、請求體等。
- 點擊“開始測試”后，工具會并發發送指定次數的請求，并實時展示結果。
總結
通過 PyQt5 和 asyncio，我們成功實現了一個美觀實用的接口并發測試工具。在這個項目中，測試工程師可以直觀地配置接口參數并分析響應結果，同時也能深入理解 Python 的異步編程原理。
趕緊試試吧！
更新公告)
