簡介

WebSocket 是一種在單個(gè) TCP 連接上進(jìn)行全雙工通信(Full Duplex 是通訊傳輸?shù)囊粋€(gè)術(shù)語。通信允許數(shù) 據(jù)在兩個(gè)方向上同時(shí)傳輸，它在能力上相當(dāng)于兩個(gè)單工通信方式的結(jié)合。全雙工指可以同時(shí)（瞬時(shí)）進(jìn) 行信號(hào)的雙向傳輸（ A→B 且 B→A ）。指 A→B 的同時(shí) B→A，是瞬時(shí)同步的)的協(xié)議。

WebSocket 通信協(xié)議于 2011 年被 IETF 定為標(biāo)準(zhǔn) RFC 6455，并由 RFC7936 補(bǔ)充規(guī)范。WebSocket API (WebSocket API 是一個(gè)使用WebSocket 協(xié)議的接口，通過它來建立全雙工通道來收發(fā)消息) 也被 W3C 定為標(biāo)準(zhǔn)。

WebSocket 使得客戶端和服務(wù)器之間的數(shù)據(jù)交換變得更加簡單，允許服務(wù)端主動(dòng)向客戶端推送數(shù)據(jù)。 在 WebSocket API 中，瀏覽器和服務(wù)器只需要完成一次握手，兩者之間就直接可以創(chuàng)建持久性的連接， 并進(jìn)行雙向數(shù)據(jù)傳輸。

而 HTTP 協(xié)議就不支持持久連接，雖然在 HTTP1.1 中進(jìn)行了改進(jìn)，使得有一個(gè) keep-alive，在一個(gè) HTTP 連接中，可以發(fā)送多個(gè) Request，接收多個(gè) Response。

但是在 HTTP 中 Request = Response 永遠(yuǎn)是成立的，也就是說一個(gè) request 只能有一個(gè)response。而且 這個(gè)response也是被動(dòng)的，不能主動(dòng)發(fā)起。

websocket 常用于社交/訂閱、多玩家游戲、協(xié)同辦公/編輯、股市基金報(bào)價(jià)、體育實(shí)況播放、音視頻聊 天/視頻會(huì)議/在線教育、智能家居與基于位置的應(yīng)用。

websocket 接口不能使用 requests 直接進(jìn)行接口的調(diào)用，可以依賴第三方庫的方式來實(shí)現(xiàn)調(diào)用，以下內(nèi) 容介紹如何調(diào)用第三方庫實(shí)現(xiàn) websocket 的接口自動(dòng)化測試。

實(shí)戰(zhàn)

使用 python 語言實(shí)現(xiàn) websocket 的接口自動(dòng)化

環(huán)境準(zhǔn)備

1. 安裝 pyhton3 環(huán)境
2. 下載需要的運(yùn)行庫

pip install websocket-client

實(shí)戰(zhàn)演示

1.連接 websoket 服務(wù)器

import logging
from websocket import create_connection
logger = logging.getLogger(__name__)
url = 'ws://echo.websocket.org/' #一個(gè)在線的回環(huán)websocket接口,必須以websocket的方式連
接后訪問，無法直接在網(wǎng)頁端輸入該地址訪問
wss = create_connection(url, timeout=timeout)

1.發(fā)送websoket消息

wss.send('Hello World')

1.接收websocket消息

res = wss.recv()
iogger.info(res)

1.關(guān)閉websocket連接

wss.close()

1.websocket 第三方庫的調(diào)用不支持直接發(fā)送除字符串外的其他數(shù)據(jù)類型，所以在發(fā)送請求之前需要將 Python 結(jié)構(gòu)化的格式,轉(zhuǎn)換為成為字符串類型或者 json 字符串后，再發(fā)起 websocket 的接口請求

#待發(fā)送的數(shù)據(jù)體格式為：
data= {
 "a" : "abcd",
 "b" : 123
 }
# 發(fā)送前需要把數(shù)據(jù)處理成 json 字符串
new_data=json.dumps(data,ensure_ascii=False)
wss.send(new_data)

1.接收的數(shù)據(jù)體的處理： 如果接口定義為 json 的話，由于數(shù)據(jù)的傳輸都是字符串格式的，需要對(duì)接收 的數(shù)據(jù)體進(jìn)行轉(zhuǎn)換操作

# 接收的數(shù)據(jù)體的格式也為字符串
logger.info(type(res)) # <class 'str'>

對(duì)于響應(yīng)內(nèi)容進(jìn)行格式轉(zhuǎn)換處理：

def load_json(base_str):
 if isinstance(base_str, str):
 try:
 res = json.loads(base_str)
 return load_json(res)
 except JSONDecodeError:
 return base_str
 elif isinstance(base_str, list):
 res = []
 for i in base_str:
 res.append(load_json(i))
 return res
 elif isinstance(base_str, dict):
 for key, value in base_str.items():
 base_str[key] = load_json(value)
 return base_str
 return base_str

1.websocket接口自動(dòng)化測試，二次封裝demo展示

web_socket_util.py封裝websocket接口通用操作：

import logging
import json
from websocket import create_connection
logger = logging.getLogger(__name__)
class WebsocketUtil():
 def conn(self, uri, timeout=3):
 '''
 連接web服務(wù)器
 :param uri: 服務(wù)的url
 :param timeout: 超時(shí)時(shí)間
 :return:
 '''
 self.wss = create_connection(uri, timeout=timeout)
 def send(self, message):
 '''
 發(fā)送請求數(shù)據(jù)體
 :param message: 待發(fā)送的數(shù)據(jù)信息
 :return:
 '''
 if not isinstance(message, str):
 message = json.dumps(message)
 return self.wss.send(message)
 def load_json(self, base_str):
 '''
 進(jìn)行數(shù)據(jù)體的處理
 :param base_str: 待處理的數(shù)據(jù)體
 :return:
 '''
 if isinstance(base_str, str):
 try:
 res = json.loads(base_str)
 return self.load_json(res)
 except JSONDecodeError:
 return base_str
 elif isinstance(base_str, list):
 res = []
 for i in base_str:
 res.append(self.load_json(i))
 return res
 elif isinstance(base_str, dict):
 for key, value in base_str.items():
 base_str[key] = self.load_json(value)
 return base_str
 return base_str
 def recv(self, timeout=3):
 '''
 接收數(shù)據(jù)體信息，并調(diào)用數(shù)據(jù)體處理方法處理響應(yīng)體
 :param timeout: 超時(shí)時(shí)間
 :return:
 '''
 if isinstance(timeout, dict):
 timeout = timeout["timeout"]
 try:
 self.settimeout(timeout)
 recv_json = self.wss.recv()
 all_json_recv = self.load_json(recv_json)
 self._set_response(all_json_recv)
 return all_json_recv
 except WebSocketTimeoutException:
 logger.error(f"已經(jīng)超過{timeout}秒沒有接收數(shù)據(jù)啦")
def settimeout(self, timeout):
 '''
 設(shè)置超時(shí)時(shí)間
 :param timeout: 超時(shí)時(shí)間
 :return:
 '''
 self.wss.settimeout(timeout)
 def recv_all(self, timeout=3):
 '''
 接收多個(gè)數(shù)據(jù)體信息，并調(diào)用數(shù)據(jù)體處理方法處理響應(yīng)體
 :param timeout: 超時(shí)時(shí)間
 :return:
 '''
 if isinstance(timeout, dict):
 timeout = timeout["timeout"]
 recv_list = []
 while True:
 try:
 self.settimeout(timeout)
 recv_json = self.wss.recv()
 all_json_recv = self.load_json(recv_json)
 recv_list.append(all_json_recv)
 logger.info(f"all::::: {all_json_recv}")
 except WebSocketTimeoutException:
 logger.error(f"已經(jīng)超過{timeout}秒沒有接收數(shù)據(jù)啦")
 break
 self._set_response(recv_list)
 return recv_list
 def close(self):
 '''
 關(guān)閉連接
 :return:
 '''
 return self.wss.close()
 def _set_response(self, response):
 self.response = response
 def _get_response(self) -> list:
 return self.response

test_case.py websocket接口自動(dòng)化測試用例：文章來源地址http://www.zghlxwxcb.cn/news/detail-520225.html

class TestWsDemo:
 def setup(self):
 url = 'ws://echo.websocket.org/'
 self.wss = WebsocketUtil()
 self.wss.conn(url)
 def teardown(self):
 self.wss.close()
 def test_demo(self):
 data = {"a": "hello", "b": "world"}
 self.wss.send(data)
 res = self.wss.recv()
 assert 'hello' == res['a']

到了這里，關(guān)于軟件測試 | websocket接口測試的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！