Python WebSocket接口测试实战：从环境搭建到长短连接实现

季宁

发布时间：2026-10-14| 7人浏览过

原创

本文详解如何使用Python进行WebSocket接口测试。内容涵盖websocket-client库的安装、通过HTTP代理连接、核心回调方法（on_message、on_error、on_open、on_close）的配置，以及长连接（run_forever）与短连接（create_connection）的具体代码实现与区别分析。

Python WebSocket测试概述

在进行接口测试时，除了常规的HTTP接口外，Socket接口也是常见类型之一。本文将重点介绍如何利用Python对WebSocket接口进行测试。

环境安装与基础连接

鉴于当前主流应用多采用WebSocket协议，我们首先需完成相关测试包的安装工作。


pip install websocket-client

完成环境配置后，即可正式进入WebSocket测试环节。

首先通过一个基础示例进行演示：


import websocket
ws = websocket.WebSocket()
ws.connect("ws://example.com/websocket", 
      http_proxy_host="proxy_host_name", 
      http_proxy_port=3128)

该示例展示了如何建立WebSocket连接，该模块支持通过HTTP代理访问WebSocket。代理服务器允许使用connect方法连接到WebSocket端口。默认的squid设置是“只允许连接HTTPS端口”。

核心回调方法详解

在WebSocket测试中，主要涉及以下几个关键方法：

on_message方法：


def on_message(ws, message):
  print(message)

on_message用于接收消息，服务器发送的所有消息均可通过此方法接收。

on_error方法：


def on_error(ws, error):
  print(error)

该方法用于处理错误异常。一旦Socket程序出现通信问题，即可通过此方法捕获。

on_open方法：


def on_open(ws):
  def run(*args):
    for i in range(30):
      # send the message, then wait
      # so thread doesn't exit and socket
      # isn't closed
      ws.send("Hello %d" % i)
      time.sleep(1)

    time.sleep(1)
    ws.close()
    print("Thread terminating...")

  Thread(target=run).start()

on_open方法用于保持连接。上述示例展示了保持连接的过程，每隔一段时间执行一次操作，它会在30s内一直发送hello。最后停止。

on_close方法：


def on_close(ws):
  print("### closed ###")

onclose主要用于关闭Socket连接。

WebSocket应用创建与连接管理

如何创建一个WebSocket应用：


ws = websocket.WebSocketApp("wss://echo.websocket.org")

括号内指定了要连接的Socket地址。在WebSocketApp实例化方法中，还可以传入其他参数，这些参数即为我们前述介绍的方法。


ws = websocket.WebSocketApp("ws://echo.websocket.org/",
              on_message=on_message,
              on_error=on_error,
              on_close=on_close)

指定这些参数后，即可直接调用，例如：


ws.on_open = on_open

上述代码即调用了on_open方法。

若希望Socket保持长连接，持续保持连接状态，可使用run_forever方法：


ws.run_forever()

完整代码示例如下：


import websocket
from threading import Thread
import time
import sys

def on_message(ws, message):
  print(message)

def on_error(ws, error):
  print(error)

def on_close(ws):
  print("### closed ###")

def on_open(ws):
  def run(*args):
    for i in range(3):
      # send the message, then wait
      # so thread doesn't exit and socket
      # isn't closed
      ws.send("Hello %d" % i)
      time.sleep(1)

    time.sleep(1)
    ws.close()
    print("Thread terminating...")

  Thread(target=run).start()


if __name__ == "__main__":

  websocket.enableTrace(True)
  host = "ws://echo.websocket.org/"
  ws = websocket.WebSocketApp(host,
                on_message=on_message,
                on_error=on_error,
                on_close=on_close)
  ws.on_open = on_open
  ws.run_forever()

若需发送一条短消息并在完成后立即断开连接，可使用短连接方式：


from websocket import create_connection
ws = create_connection("ws://echo.websocket.org/")
print("Sending 'Hello, World'...")
ws.send("Hello, World")
print("Sent")
print("Receiving...")
result = ws.recv()
print("Received '%s'" % result)
ws.close()

关于WebSocket的介绍至此结束。

以上是用Python进行WebSocket接口测试的详细内容，更多关于Python接口测试的资料请关注本站其他相关文章！