• websocket接口测试

    参考文章：websocket接口自动化集成pytest测试框架

    对比需要学习：轮询、长轮询、websocket  三者关系 

    1、工作中遇到的情况

    F12功能看到的数据

     

      

    2、python中操作ws

    （1）websocket包

    安装

    pip install websocket -i https://pypi.douban.com/simple/

    参考文章：你真的了解WebSocket吗？ - 武沛齐 - 博客园

    参考视频：08 python fullstack s9day131 websocket原理剖析_哔哩哔哩_bilibili

    非常详细，就是照抄学习

    （2）flask-websocket

    参考文章：Flask教程(十九)SocketIO - 迷途小书童的Note迷途小书童的Note

    参考视频：Flask Web开发教程(十九)SocketIO_哔哩哔哩_bilibili

    代码：index.html

    1. html>
    2. <html lang="en">
    3. <head>
    4.     <meta charset="UTF-8">
    5.     <title>SocketIO Demotitle>
    6.     <script type="text/javascript" src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js">script>
    8.     <script type="text/javascript" src="//cdnjs.cloudflare.com/ajax/libs/socket.io/4.5.0/socket.io.js">script>
    9. head>
    10. <body>
    11.  
    12. <h2>Demo of SocketIOh2>
    13. <div id="t">div>
    14. <script>
    15. $(document).ready(function () {
    16.     namespace = '/dcenter';
    17.     var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace);
    18.     socket.on('dcenter', function (res) {
    19.         var t = res.data;
    20.         if (t) {
    21.             $("#t").append(t).append('
      '      );
    22.         }
    23.  
    24.     });
    25. });
    26. script>
    27. body>
    28. html>

     fw.py

    1. # -*- coding:utf-8 -*-
    2. """
    3. 基于flask开发websocket服务
    4. 使用的包是flask-socketio
    5. """
    6.  
    7. from flask import Flask, render_template
    8. from flask_socketio import SocketIO, emit
    9.  
    10. app = Flask(__name__)
    11. app.config['SECRET_KEY'] = 'secret_key'
    12.  
    13. socketio = SocketIO()
    14. socketio.init_app(app, cors_allowed_origins='*')
    15.  
    16. name_space = '/dcenter'
    17.  
    18.  
    19. @app.route('/')
    20. def index():
    21.     return render_template('index1.html')
    22.  
    23.  
    24. @app.route('/push')
    25. def push_once():
    26.     event_name = 'dcenter'
    27.     broadcasted_data = {'data': "test message!"}
    28.     socketio.emit(event_name, broadcasted_data, broadcast=False, namespace=name_space)
    29.     return 'done!'
    30.  
    31.  
    32. @socketio.on('connect', namespace=name_space)
    33. def connected_msg():
    34.     print('client connected.')
    35.  
    36.  
    37. @socketio.on('disconnect', namespace=name_space)
    38. def disconnect_msg():
    39.     print('client disconnected.')
    40.  
    41.  
    42. @socketio.on('my_event', namespace=name_space)
    43. def mtest_message(message):
    44.     print(message)
    45.     emit('my_response',
    46.          {'data': message['data'], 'count': 1})
    47.  
    48.  
    49. if __name__ == '__main__':
    50.     socketio.run(app, host='127.0.0.1', port=5000, debug=True)

    安装

    python39 -m pip install --upgrade Flask-SocketIO                升级到了最高版本

    python39 -m pip install --upgrade python-socketio==4.6.0   升级到指定版本

    python39 -m pip install python-engineio==3.13.2                  安装指定版本

    实际操作版本介绍：

    eventlet                       0.33.1

    python-engineio           4.3.4
    python-socketio           5.7.2
    Flask                           2.2.2
    Flask-SocketIO           5.3.1
    Python                        3.9.4

     代码与参考文章一致

    重点：！！！！！！！！！！！！！！！！！！！！！！！！

    下图问题原因：

    The client is using an unsupported version of the Socket.IO or Engine.IO protocols

    python - The client is using an unsupported version of the Socket.IO or Engine.IO protocols Error - Stack Overflow

    版本不匹配的原因，下面的链接找到的答案：方式就是根据socket.io版本降低，或者升高socket.io的版本

    在python-socketio的官网有说明：https://pypi.org/project/python-socketio/

     根据我安装的python-socketio的版本升高js的socket.io版本

    //cdnjs.cloudflare.com/ajax/libs/socket.io/4.5.0/socket.io.js

    3、ws接口mock

    实际使用操作一波

    4、ws接口测试

    安装

    pip install websocket -i https://pypi.douban.com/simple/

    pip install websocket-client

    clinet.py

    1. # -*- coding:utf-8 -*-
    2.  
    3. class websocketclient:
    4.     def __init__(self):
    5.         self.host = 'wss://url'
    6.  
    7.     def send(self, params):
    8.         try:
    9.             self.ws.send(params)
    10.             print(f'发送数据成功：{params}')
    11.         except Exception as e:
    12.             print(f'发送数据{params}失败')
    13.  
    14.     def recv(self):
    15.         try:
    16.             res = self.ws.recv()
    17.             print(f'接收数据成功：{res}')
    18.             return res
    19.         except Exception as e:
    20.             print(f'接收数据{res}失败')
    21.             return ''

    imserver_api.py

    1. # -*- coding:utf-8 -*-
    2.  
    3. import websocket
    4. from nomalstudy.client import websocketclient
    5.  
    6.  
    7. class ImServerApi(websocketclient):
    8.     def __init__(self, timeout=5):
    9.         super(ImServerApi, self).__init__()
    10.         self.url = f'{self.host}/urgeSocket?_token=6t6OGmrt3xm1SYqDRFrUUeHVhLvHvMVKJQ3UBxaQ4kK9RMde&_appType=receive'
    11.         self.ws = websocket.create_connection(self.url, timeout=timeout)

    test_websocket_api.py

    1. # -*- coding:utf-8 -*-
    2.  
    3. import json
    4. import pytest
    5. from nomalstudy.imserver_api import ImServerApi
    6.  
    7.  
    8. class TestImServerApi:
    9.     kfid = ''  # 定义客服id，全局变量作为各个测试用例的关联数据
    10.  
    11.     def setup_class(self):
    12.         self.im = ImServerApi()  # 创建一个websocket协议的接口对象
    13.  
    14.     # 测试客服匹配
    15.     def test_match(self):
    16.         params = {
    17.             "msgId": "111",
    18.             "type": "match",
    19.             "from": "shamo",
    20.             "to": "system"
    21.         }
    22.         self.im.send('heartbeat')
    23.         res = self.im.recv()
    24.         assert res == 'heartbeat'
    25.         # res = json.loads(res)  # 将其转换成json对象
    26.         # assert res['code'] == '0'
    27.         # 提取msg，msg是匹配到的客服id
    28.         # self.__class__.kfid = res['msg']
    29.  
    30.     # 测试给客服发送正常消息
    31.     def test_message(self):
    32.         params = {
    33.             "msgId": "111",
    34.             "type": "normal",
    35.             "from": "admin",
    36.             "to": f"{self.__class__.kfid}",
    37.             "msg": "你好"
    38.         }
    39.         self.im.send(json.dumps(params))
    40.         res = self.im.recv()
    41.         res = json.loads(res)  # 将其转换成json对象
    42.         pytest.assume(res['code'] == '0', f'期望值是0,实际结果是{res["code"]}')
    43.         pytest.assume(res['msg'] == 'push success', f'期望值是0,实际结果是{res["msg"]}')
    44.         # 再次接收客服发来的数据
    45.         res = self.im.recv()
    46.         res = json.loads(res)  # 将其转换成json对象
    47.         pytest.assume(res['code'] == '0', f'期望值是0,实际结果是{res["code"]}')
    48.         pytest.assume(res['msg'] == '同学，你好，非常高兴为你服务，有什么需要我帮忙的呢?', f'期望值是0,实际结果是{res["msg"]}')
    49.  
    50.     # 测试发送数据时消息是空的
    51.     def test_message_msgisnull(self):
    52.         params = {
    53.             "msgId": "111",
    54.             "type": "normal",
    55.             "from": "admin",
    56.             "to": f"{self.kfid}",
    57.             "msg": ""
    58.         }
    59.         self.im.send(json.dumps(params))
    60.         res = self.im.recv()
    61.         res = json.loads(res)  # 将其转换成json对象
    62.         # 断言系统推送消息时对于消息的判断
    63.         pytest.assume(res['code'] == '1', f'期望值是1,实际结果是{res["code"]}')
    64.         pytest.assume(res['msg'] == '消息内容为空', f'期望值是0,实际结果是{res["msg"]}')

  • 相关阅读:
    从底层结构开始学习FPGA（6）----分布式RAM（DRAM，Distributed RAM）
    Vue3.2+TypeScript管理模块记录1-登录模块
    Python中的HTTP高手：如何玩转requests模块
    2022年全球光纤连接市场将达50.1亿美元
    首饰饰品经营商城小程序的作用是什么
    ONNX OpenVino TensorRT MediaPipe NCNN Diffusers
    API经济下新物种的诞生
    服务老是被攻击，如何设计一套比较安全的接口访问策略？
    MySQL中的事务
    BUUCTF 秘密文件 1
  • 原文地址：https://blog.csdn.net/chenchen_nini/article/details/127534427