1.連線websocket伺服器

import logging
from websocket import creat_connertion
logger = logging.getLogger(__name__)
url = 'ws://echo.websocket.org' #一個線上迴環websocket介面，必須以websocket的方式連線後訪問，無法直接在網頁端輸入該地址訪問
wss = creat_connertion(url, timeout=timeout)

2.傳送websocket訊息

wss.send('hello world')

3.接受websocket訊息

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

4.關閉websocket連線

wss.close()

websocket第三方庫的呼叫不支援直接傳送除字串外的其他資料型別，所以在傳送請求之前需要將python結構化的格式，轉換成字串型別或者json字串後，再發起websocket的介面請求。

#待發送的資料體格式為：
data= {
      "a" : "abcd",
      "b" : 123
}
#傳送前需要把資料處理成json字串
new_data = json.dumps(data,ensure_ascii=Flase)
wss.send(new_data)

接收的資料體的處理：如果介面定義為json的話，由於資料的傳輸都是字串格式的，需要對接收的資料進行轉換操作

#接收的資料體的格式也是字串：
logger.info(type(res)) #<class 'str'>

對於響應內容進行格式轉換處理：

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,str):
        for key, value in base_str.items():
            base_str[key]=load_json(value)
        return base_str
    return base_str

websocket介面自動化測試，二次封裝demo展示

web_socket_util.py

import logging
import json
from json import JSONDecodeError

from websocket import create_connection

logger = logging.getLogger(__name__)


class WebsocketUtil():
    def conn(self, uri, timeout=3):
        '''
        連線web伺服器
        :param uri: 服務的url
        :param timeout: 超時時間
        :return:
        '''
        self.wss = create_connection(uri, timeout=timeout)

    def send(self, message):
        '''
        傳送請求資料體
        :param message: 待發送的資料資訊
        :return:
        '''
        if not isinstance(message, str):
            message = json.dumps(message)
        return self.wss.send(message)

    def load_json(self, base_str):
        '''
        進行資料體處理
        :param base_str: 待處理的資料
        :return:
        '''
        if isinstance(base_str, str):
            try:
                res = json.loads(base_str)
                return base_str
            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, str):
            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):
        '''
        接收資料體資訊，並呼叫資料體處理方法處理響應體
        :param timeout: 超時時間
        :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'已經超過{timeout}秒沒有接收資料啦')

    def settimeout(self, timeout):
        '''
        設定超時時間
        :param timeout: 超時時間
        :return:
        '''
        self.wss.settimeout(timeout)

    def recv_all(self, timeout=3):
        '''
        姐搜多個數據體資訊，並呼叫資料體處理方法處理響應體
        :param timeout: 超時時間
        :return:
        '''
        if isinstance(timeout, dict):
            timeout = timeout['timeout']
        recv_list = []
        while True:
            try:
                self.settimeout(timeout)
                recv_list = self.wss.recv()
                all_json_recv = self.load_json(recv_list)
                recv_list.append(all_json_recv)
                logger.info(f'all::::: {all_json_recv}')
            except WebSocketTimeoutException:
                logger.error(f'已經超過{timeout}秒沒有接收資料啦')
                break
        self._set_response(recv_list)
        return recv_list

    def close(self):
        '''
        關閉連線
        :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介面自動化測試用例：

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']