1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
"""Test serialize/deserialize messages with buffers"""
import os
from jupyter_client.session import Session
from jupyter_server.services.kernels.connection.base import (
deserialize_binary_message,
serialize_binary_message,
)
def test_serialize_binary():
s = Session()
msg = s.msg("data_pub", content={"a": "b"})
msg["buffers"] = [memoryview(os.urandom(3)) for i in range(3)]
bmsg = serialize_binary_message(msg)
assert isinstance(bmsg, bytes)
def test_deserialize_binary():
s = Session()
msg = s.msg("data_pub", content={"a": "b"})
msg["buffers"] = [memoryview(os.urandom(2)) for i in range(3)]
bmsg = serialize_binary_message(msg)
msg2 = deserialize_binary_message(bmsg)
assert msg2 == msg
|
