国产成人精品亚洲777人妖,欧美日韩精品一区视频,最新亚洲国产,国产乱码精品一区二区亚洲

您的位置:首頁技術文章
文章詳情頁

用Python進行websocket接口測試

瀏覽:4日期:2022-07-08 09:09:31

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

現在大多數用的都是websocket,那我們就先來安裝一下websocket的安裝包。

pip install websocket-client

用Python進行websocket接口測試

安裝完之后,我們就開始我們的websocket之旅了。

我們先來看個炒雞簡單的栗子:

import websocketws = 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是用來接受消息的,server發送的所有消息都可以用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應用:

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 websocketfrom threading import Threadimport timeimport sysdef 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_connectionws = 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 接口測試的資料請關注好吧啦網其它相關文章!

標簽: Python 編程
相關文章:
主站蜘蛛池模板: 定西市| 肇庆市| 池州市| 齐齐哈尔市| 宁阳县| 凯里市| 大同市| 彭山县| 平顶山市| 洛川县| 青岛市| 鄂尔多斯市| 庆阳市| 延津县| 开封县| 仁寿县| 五大连池市| 富锦市| 高密市| 虹口区| 屯昌县| 阿坝县| 丰城市| 阿巴嘎旗| 丹东市| 卢湾区| 大安市| 井研县| 莫力| 蓝山县| 拜城县| 通州市| 辽中县| 襄垣县| 商南县| 雷山县| 崇仁县| 夏河县| 天镇县| 左贡县| 曲阳县|