做网站需求报告,wordpress 展示类主题,wordpress搭建it博客,app制作开发费用参考文章#xff1a;websocket接口自动化集成pytest测试框架
对比需要学习#xff1a;轮询、长轮询、websocket 三者关系
1、工作中遇到的情况
F12功能看到的数据 2、python中操作ws
#xff08;1#xff09;websocket包
安装 pip install websocket -i https://pyp…参考文章websocket接口自动化集成pytest测试框架
对比需要学习轮询、长轮询、websocket 三者关系
1、工作中遇到的情况
F12功能看到的数据 2、python中操作ws
1websocket包
安装 pip install websocket -i https://pypi.douban.com/simple/ 参考文章你真的了解WebSocket吗 - 武沛齐 - 博客园
参考视频08 python fullstack s9day131 websocket原理剖析_哔哩哔哩_bilibili
非常详细就是照抄学习
2flask-websocket
参考文章Flask教程(十九)SocketIO - 迷途小书童的Note迷途小书童的Note
参考视频Flask Web开发教程(十九)SocketIO_哔哩哔哩_bilibili
代码index.html
!DOCTYPE html
html langen
headmeta charsetUTF-8titleSocketIO Demo/titlescript typetext/javascript src//cdn.bootcss.com/jquery/3.1.1/jquery.min.js/script
!-- script typetext/javascript src//cdn.bootcss.com/socket.io/1.5.1/socket.io.min.js/script--script typetext/javascript src//cdnjs.cloudflare.com/ajax/libs/socket.io/4.5.0/socket.io.js/script
/head
bodyh2Demo of SocketIO/h2
div idt/div
script
$(document).ready(function () {namespace /dcenter;var socket io.connect(location.protocol // document.domain : location.port namespace);socket.on(dcenter, function (res) {var t res.data;if (t) {$(#t).append(t).append(br/);}});
});
/script
/body
/html fw.py
# -*- coding:utf-8 -*-基于flask开发websocket服务
使用的包是flask-socketio
from flask import Flask, render_template
from flask_socketio import SocketIO, emitapp Flask(__name__)
app.config[SECRET_KEY] secret_keysocketio SocketIO()
socketio.init_app(app, cors_allowed_origins*)name_space /dcenterapp.route(/)
def index():return render_template(index1.html)app.route(/push)
def push_once():event_name dcenterbroadcasted_data {data: test message!}socketio.emit(event_name, broadcasted_data, broadcastFalse, namespacename_space)return done!socketio.on(connect, namespacename_space)
def connected_msg():print(client connected.)socketio.on(disconnect, namespacename_space)
def disconnect_msg():print(client disconnected.)socketio.on(my_event, namespacename_space)
def mtest_message(message):print(message)emit(my_response,{data: message[data], count: 1})if __name__ __main__:socketio.run(app, host127.0.0.1, port5000, debugTrue)安装 python39 -m pip install --upgrade Flask-SocketIO 升级到了最高版本 python39 -m pip install --upgrade python-socketio4.6.0 升级到指定版本 python39 -m pip install python-engineio3.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
# -*- coding:utf-8 -*-class websocketclient:def __init__(self):self.host wss://urldef send(self, params):try:self.ws.send(params)print(f发送数据成功{params})except Exception as e:print(f发送数据{params}失败)def recv(self):try:res self.ws.recv()print(f接收数据成功{res})return resexcept Exception as e:print(f接收数据{res}失败)return
imserver_api.py
# -*- coding:utf-8 -*-import websocket
from nomalstudy.client import websocketclientclass ImServerApi(websocketclient):def __init__(self, timeout5):super(ImServerApi, self).__init__()self.url f{self.host}/urgeSocket?_token6t6OGmrt3xm1SYqDRFrUUeHVhLvHvMVKJQ3UBxaQ4kK9RMde_appTypereceiveself.ws websocket.create_connection(self.url, timeouttimeout)test_websocket_api.py
# -*- coding:utf-8 -*-import json
import pytest
from nomalstudy.imserver_api import ImServerApiclass TestImServerApi:kfid # 定义客服id全局变量作为各个测试用例的关联数据def setup_class(self):self.im ImServerApi() # 创建一个websocket协议的接口对象# 测试客服匹配def test_match(self):params {msgId: 111,type: match,from: shamo,to: system}self.im.send(heartbeat)res self.im.recv()assert res heartbeat# res json.loads(res) # 将其转换成json对象# assert res[code] 0# 提取msgmsg是匹配到的客服id# self.__class__.kfid res[msg]# 测试给客服发送正常消息def test_message(self):params {msgId: 111,type: normal,from: admin,to: f{self.__class__.kfid},msg: 你好}self.im.send(json.dumps(params))res self.im.recv()res json.loads(res) # 将其转换成json对象pytest.assume(res[code] 0, f期望值是0,实际结果是{res[code]})pytest.assume(res[msg] push success, f期望值是0,实际结果是{res[msg]})# 再次接收客服发来的数据res self.im.recv()res json.loads(res) # 将其转换成json对象pytest.assume(res[code] 0, f期望值是0,实际结果是{res[code]})pytest.assume(res[msg] 同学你好非常高兴为你服务有什么需要我帮忙的呢?, f期望值是0,实际结果是{res[msg]})# 测试发送数据时消息是空的def test_message_msgisnull(self):params {msgId: 111,type: normal,from: admin,to: f{self.kfid},msg: }self.im.send(json.dumps(params))res self.im.recv()res json.loads(res) # 将其转换成json对象# 断言系统推送消息时对于消息的判断pytest.assume(res[code] 1, f期望值是1,实际结果是{res[code]})pytest.assume(res[msg] 消息内容为空, f期望值是0,实际结果是{res[msg]})
