File: test_serialize.py

package info (click to toggle)
jupyter-server 1.2.2-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, sid
  • size: 1,844 kB
  • sloc: python: 12,331; makefile: 175; javascript: 72
file content (25 lines) | stat: -rw-r--r-- 736 bytes parent folder | download
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
"""Test serialize/deserialize messages with buffers"""

import os

from jupyter_client.session import Session
from jupyter_server.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'] = [ 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