快捷導(dǎo)航

用Python進(jìn)行websocket接口測試

更新時間：2020年10月16日 15:05:55 作者：我被狗咬了

這篇文章主要介紹了用Python進(jìn)行websocket接口測試，幫助大家更好的理解和使用python，感興趣的朋友可以了解下

我們在做接口測試時，除了常見的http接口，還有一種比較多見，就是socket接口，今天講解下怎么用Python進(jìn)行websocket接口測試。

現(xiàn)在大多數(shù)用的都是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)

這個栗子就是創(chuàng)建一個websocket連接，這個模塊支持通過http代理訪問websocket。代理服務(wù)器允許使用connect方法連接到websocket端口。默認(rèn)的squid設(shè)置是“只允許連接HTTPS端口”。

在websocket里，我們有常用的這幾個方法：

on_message方法：

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

on_message是用來接受消息的，server發(fā)送的所有消息都可以用on_message這個方法來收取。

on_error方法：

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

這個方法是用來處理錯誤異常的，如果一旦socket的程序出現(xiàn)了通信的問題，就可以被這個方法捕捉到。

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內(nèi)一直發(fā)送hello。最后停止。

on_close方法：

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

onclose主要就是關(guān)閉socket連接的。

如何創(chuàng)建一個websocket應(yīng)用：

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

括號里面就是你要連接的socket的地址，在WebSocketApp這個實例化的方法里面還可以有其他參數(shù)，這些參數(shù)就是我們剛剛介紹的這些方法。

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

指定了這些參數(shù)之后就可以直接進(jìn)行調(diào)用了，例如：

ws.on_open = on_open

這樣就是調(diào)用了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()

關(guān)于websocket的介紹就到這兒了。

以上就是用Python進(jìn)行websocket接口測試的詳細(xì)內(nèi)容，更多關(guān)于python 接口測試的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: