WebSocket 接口测试浅谈

什么是 WebSocket

WebSocket 是一种基于在单个 TCP 连接上进行全双工通信的协议,解决了HTTP协议不适用于实时通信的缺点,相较于 HTTP 协议,WebSocket 协议实现了持久化网络通信,可以实现客户端和服务端的长连接,能够进行双向实时通信,协议名为"ws"。

WebSocket 使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。在 WebSocket API 中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。
在 WebSocket API 中,浏览器和服务器只需要做一个握手的动作,然后,浏览器和服务器之间就形成了一条快速通道。两者之间就直接可以数据互相传送。

HTTP与WebSocket

image-1662173780431

WebSocket的特点

  • 建立在 TCP 协议之上,服务器端的实现比较容易。
  • 与 HTTP 协议有着良好的兼容性。默认端口也是80和443,并且握手阶段采用 HTTP 协议,因此握手时不容易屏蔽,能通过各种 HTTP 代理服务器。
  • 数据格式比较轻量,性能开销小,通信高效。
  • 可以发送文本,也可以发送二进制数据。
  • 没有同源限制,客户端可以与任意服务器通信。
  • 协议标识符是ws(如果加密,则为wss),服务器网址就是 URL。
ws://example.com:80/some/path

image-1662191779893

WebSocket测试方法

在线测试工具

http://www.jsons.cn/websocket/

使用python 编程作为客户端测试

安装

Install with pip

pip install websocket-server

使用接口进行通信

我们使用python写一个简单的websocket的服务端

#! -*- coding: utf-8 -*-
"""
Info: Websocket 的使用示例
"""
import asyncio
import websockets

websocket_users = set()


# 检测客户端权限,用户名密码通过才能退出循环
async def check_user_permit(websocket):
    print("new websocket_users:", websocket)
    websocket_users.add(websocket)
    print("websocket_users list:", websocket_users)
    while True:
        recv_str = await websocket.recv()
        cred_dict = recv_str.split(":")
        if cred_dict[0] == "admin" and cred_dict[1] == "123456":
            response_str = "Congratulation, you have connect with server..."
            await websocket.send(response_str)
            print(cred_dict)
            print("Password is ok...")
            return True
        else:
            print(cred_dict)
            response_str = "Sorry, please input the username or password..."
            print("Password is wrong...")
            await websocket.send(response_str)


# 接收客户端消息并处理,这里只是简单把客户端发来的返回回去
async def recv_user_msg(websocket):
    while True:
        recv_text = await websocket.recv()
        print("recv_text:", websocket.pong, recv_text)
        response_text = f"Server return: {recv_text}"
        print("response_text:", response_text)
        await websocket.send(response_text)


# 服务器端主逻辑
async def run(websocket, path):
    while True:
        try:
            await check_user_permit(websocket)
            await recv_user_msg(websocket)
        except websockets.ConnectionClosed:
            print("ConnectionClosed...", path)    # 链接断开
            print("websocket_users old:", websocket_users)
            websocket_users.remove(websocket)
            print("websocket_users new:", websocket_users)
            break
        except websockets.InvalidState:
            print("InvalidState...")    # 无效状态
            break
        except Exception as e:
            print("Exception:", e)


if __name__ == '__main__':
    print("127.0.0.1:8281 websocket...")
    asyncio.get_event_loop().run_until_complete(websockets.serve(run, "127.0.0.1", 8281))
    asyncio.get_event_loop().run_forever()

运行服务端后可使用在线工具调试
image-1662256803205

使用python封装一个WebsocketUtil

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2022/9/3 16:24
# @File    : web_socket_util.py.py
# @Software: win10 Tensorflow1.13.1 python3.9
import logging
import json
from json import JSONDecodeError

from websocket import create_connection, WebSocketTimeoutException

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

编写测试用例

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2022/9/3 16:25
# @Author  : shisuiyi
# @File    : test_case.py.py
# @Software: win10 Tensorflow1.13.1 python3.9
from web_socket_util import WebsocketUtil


class TestWsDemo:
    def setup(self):
        url = 'ws://127.0.0.1:8281'
        self.wss = WebsocketUtil()
        self.wss.conn(url)
        data = 'admin:123456'
        self.wss.send(data)
        res = self.wss.recv()
        print(res)

    def teardown(self):
        self.wss.close()

    def test_send_01(self):
        data = 'shisuiyi'
        self.wss.send(data)
        res = self.wss.recv()
        print(res)
        assert data in res

image-1662256463599
image-1662256485405

上一篇 下一篇

评论 | 0条评论