test_serialize.py
26 lines
| 740 B
| text/x-python
|
PythonLexer
MinRK
|
r18335 | """Test serialize/deserialize messages with buffers""" | ||
| import os | ||||
| import nose.tools as nt | ||||
| from IPython.kernel.zmq.session import Session | ||||
| from ..base.zmqhandlers import ( | ||||
| serialize_binary_message, | ||||
| deserialize_binary_message, | ||||
| ) | ||||
| def test_serialize_binary(): | ||||
| s = Session() | ||||
| msg = s.msg('data_pub', content={'a': 'b'}) | ||||
| msg['buffers'] = [ os.urandom(3) for i in range(3) ] | ||||
| bmsg = serialize_binary_message(msg) | ||||
| nt.assert_is_instance(bmsg, bytes) | ||||
| def test_deserialize_binary(): | ||||
| s = Session() | ||||
| msg = s.msg('data_pub', content={'a': 'b'}) | ||||
| msg['buffers'] = [ os.urandom(2) for i in range(3) ] | ||||
| bmsg = serialize_binary_message(msg) | ||||
| msg2 = deserialize_binary_message(bmsg) | ||||
| nt.assert_equal(msg2, msg) | ||||