msgpack-python/test/test_bypass.py at 8b28c90096dd36ff30de589933c1c1f2b6341a90 · msgpack/msgpack-python · GitHub

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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from msgpack import Bypass, packb, unpackb


def test_roundtrip_bypassed_bytes():
    binary = b"\x00\x01\x02\x03\x04\x05"

    binary_packed = packb(binary)
    binary_bypassed_packed = packb(Bypass(binary_packed))

    assert binary_bypassed_packed == binary_packed
    assert unpackb(binary_packed) == binary


def test_roundtrip_bypassed_object():
    obj = {"key0": {"key1": {"key2": {"key3": [1, 2, 3]}}}}

    obj_packed = packb(obj)
    obj_bypassed_packed = packb(Bypass(obj_packed))

    assert obj_bypassed_packed == obj_packed
    assert unpackb(obj_bypassed_packed) == obj


def test_roundtrip_deep_bypassed_object():
    obj = {"key0": {"key1": {"key2": {"key3": [1, 2, 3]}}}}
    obj_bypassed = {"key0": {"key1": {"key2": {"key3": [1, 2, Bypass(packb(3))]}}}}

    assert packb(obj) == packb(obj_bypassed)
    assert unpackb(packb(obj)) == obj


def test_roundtrip_bypassed_object_with_cache():
    class SimpleCache:
        def __init__(self, obj):
            self.obj = obj
            self.packed = None

        def get_packed(self):
            if self.packed is None:
                self.packed = packb(self.obj)

            return Bypass(self.packed)

    def default(obj):
        if isinstance(obj, SimpleCache):
            return obj.get_packed()
        return obj

    obj = {"key0": {"key1": {"key2": {"key3": [1, 2, 3]}}}}
    cache = SimpleCache(obj)

    obj_packed = packb(cache.obj, default=default)
    cache.packed = obj_packed
    obj_bypassed_packed = packb(Bypass(cache.packed))

    assert obj_bypassed_packed == obj_packed
    assert unpackb(obj_bypassed_packed) == cache.obj