最近用了Flask-SocketIO，因为要和一些性能比较差的机器通信，所以数据格式并没有采用传统的json，而是采用Google的跨平台序列化工具FlatBuffers，它的结构化数据都以二进制形式保存，所以需要用SocketIO传输二进制格式，这个Flask-SocketIO支持的。

但在写单元测试的的时候发现Flask-SocketIO的单元测试会异常，代码和异常如下：

• test.py

from flask import Flask, render_template
from flask_socketio import SocketIO, emitapp = 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_ioclass FlaskrTestCase(unittest.TestCase):def setUp(self):self.app = appself.app.config['TESTING'] = Trueself.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):passif __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_ioself.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 emitself.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_messageraise 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 notimplemented, a callback request will just tell theserver to provide the arguments to invoke thecallback, but no callback is invoked. Instead, thearguments that the server provided for the callbackare returned by this function.:param namespace: The namespace of the event. The global namespace isassumed if this argument is not provided."""namespace = kwargs.pop('namespace', None)callback = kwargs.pop('callback', False)id = Noneif callback:self.callback_counter += 1id = self.callback_counterpkt = packet.Packet(packet.EVENT, data=[event] + list(args),namespace=namespace, id=id, binary=False)self.ack = Nonewith 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 notimplemented, a callback request will just tell theserver to provide the arguments to invoke thecallback, but no callback is invoked. Instead, thearguments that the server provided for the callbackare returned by this function.:param namespace: The namespace of the event. The global namespace isassumed if this argument is not provided."""namespace = kwargs.pop('namespace', None)callback = kwargs.pop('callback', False)binary = kwargs.pop('binary', False)id = Noneif callback:self.callback_counter += 1id = self.callback_counterpkt = packet.Packet(packet.EVENT, data=[event] + list(args),namespace=namespace, id=id, binary=binary)self.ack = Nonewith 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