最近用了Flask-SocketIO，因为要和一些性能比较差的机器通信，所以数据格式并没有采用传统的json，而是采用Google的跨平台序列化工具FlatBuffers，它的结构化数据都以二进制形式保存，所以需要用SocketIO传输二进制格式，这个Flask-SocketIO支持的。
但在写单元测试的的时候发现Flask-SocketIO的单元测试会异常，代码和异常如下：
- test.py
from flask import Flask, render_template
from flask_socketio import SocketIO, emit
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socket_io = SocketIO(binary=True)
socket_io.init_app(app)
@app.route('/')
def index():
return 'hello'
@socket_io.on('event')
def test_message(message):
print("Success: %s", str(message))
if __name__ == '__main__':
socket_io.run(app)
- test.py
import unittest
from test_app import app, socket_io
class FlaskrTestCase(unittest.TestCase):
def setUp(self):
self.app = app
self.app.config['TESTING'] = True
self.socket_io_client = socket_io.test_client(self.app)
def test_socket_io(self):
self.socket_io_client.emit('event', bytes(b'1111111'), binary=True)
def tearDown(self):
pass
if __name__ == '__main__':
unittest.main()
- 异常信息
Testing started at 5:39 PM ...
decode: encoded_packet[['51-["event",{"_placeholder":true,"num":0}]', b'1111111']]
Error
Traceback (most recent call last):
File "/mnt/hgfs/Share/fogstor-server/test.py", line 15, in test_socket_io
self.socket_io_client.emit('event', bytes(b'1111111'), binary=True)
File "/home/jeff/develop-env/fosstor_3.5/lib/python3.5/site-packages/flask_socketio/test_client.py", line 117, in emit
self.socketio.server._handle_eio_message(self.sid, pkt.encode())
File "/home/jeff/develop-env/fosstor_3.5/lib/python3.5/site-packages/socketio/server.py", line 522, in _handle_eio_message
raise ValueError('Unknown packet type.')
ValueError: Unknown packet type.
Process finished with exit code 0
这是由于Flask-SocketIO在二进制传输应用不多，所以代码并没有很好的覆盖到，只需要将
# flask_socketio文件夹下的test_client.py
......
def emit(self, event, *args, **kwargs):
"""Emit an event to the server.
:param event: The event name.
:param *args: The event arguments.
:param callback: ``True`` if the client requests a callback, ``False``
if not. Note that client-side callbacks are not
implemented, a callback request will just tell the
server to provide the arguments to invoke the
callback, but no callback is invoked. Instead, the
arguments that the server provided for the callback
are returned by this function.
:param namespace: The namespace of the event. The global namespace is
assumed if this argument is not provided.
"""
namespace = kwargs.pop('namespace', None)
callback = kwargs.pop('callback', False)
id = None
if callback:
self.callback_counter += 1
id = self.callback_counter
pkt = packet.Packet(packet.EVENT, data=[event] + list(args),
namespace=namespace, id=id, binary=False)
self.ack = None
with self.app.app_context():
self.socketio.server._handle_eio_message(self.sid, pkt.encode())
if self.ack is not None:
return self.ack['args'][0] if len(self.ack['args']) == 1 \
else self.ack['args']
......
修改为
# flask_socketio文件夹下的test_client.py
......
def emit(self, event, *args, **kwargs):
"""Emit an event to the server.
:param event: The event name.
:param *args: The event arguments.
:param callback: ``True`` if the client requests a callback, ``False``
if not. Note that client-side callbacks are not
implemented, a callback request will just tell the
server to provide the arguments to invoke the
callback, but no callback is invoked. Instead, the
arguments that the server provided for the callback
are returned by this function.
:param namespace: The namespace of the event. The global namespace is
assumed if this argument is not provided.
"""
namespace = kwargs.pop('namespace', None)
callback = kwargs.pop('callback', False)
binary = kwargs.pop('binary', False)
id = None
if callback:
self.callback_counter += 1
id = self.callback_counter
pkt = packet.Packet(packet.EVENT, data=[event] + list(args),
namespace=namespace, id=id, binary=binary)
self.ack = None
with self.app.app_context():
if binary:
encoded_packet = pkt.encode()
self.socketio.server._handle_eio_message(self.sid, encoded_packet[0])
self.socketio.server._handle_eio_message(self.sid, encoded_packet[1])
else:
self.socketio.server._handle_eio_message(self.sid, pkt.encode())
if self.ack is not None:
return self.ack['args'][0] if len(self.ack['args']) == 1 \
else self.ack['args']
......
就可以单元测试了：
Testing started at 5:40 PM ...
decode: encoded_packet[51-["event",{"_placeholder":true,"num":0}]]
reconstruct_binary: data[['event', {'_placeholder': True, 'num': 0}]], attachments[[b'1111111']]
Success: %s b'1111111'
Process finished with exit code 0
网友评论