269 lines
7.6 KiB
Python
269 lines
7.6 KiB
Python
import dataclasses
|
|
import datetime
|
|
import hashlib
|
|
import pickle
|
|
from enum import Enum
|
|
|
|
import pytest
|
|
|
|
from dbengine.serializer import TAG_TUPLE, TAG_SET, Serializer, TAG_OBJECT, TAG_ID, TAG_REF
|
|
|
|
|
|
class Obj:
|
|
def __init__(self, a, b, c):
|
|
self.a = a
|
|
self.b = b
|
|
self.c = c
|
|
|
|
def __eq__(self, other):
|
|
if id(self) == id(other):
|
|
return True
|
|
|
|
if not isinstance(other, Obj):
|
|
return False
|
|
|
|
return self.a == other.a and self.b == other.b and self.c == other.c
|
|
|
|
def __hash__(self):
|
|
return hash((self.a, self.b, self.c))
|
|
|
|
|
|
class Obj2:
|
|
class InnerClass:
|
|
def __init__(self, x):
|
|
self.x = x
|
|
|
|
def __eq__(self, other):
|
|
if not isinstance(other, Obj2.InnerClass):
|
|
return False
|
|
|
|
return self.x == other.x
|
|
|
|
def __hash__(self):
|
|
return hash(self.x)
|
|
|
|
def __init__(self, a, b, x):
|
|
self.a = a
|
|
self.b = b
|
|
self.x = Obj2.InnerClass(x)
|
|
|
|
def __eq__(self, other):
|
|
if not isinstance(other, Obj2):
|
|
return False
|
|
|
|
return (self.a == other.a and
|
|
self.b == other.b and
|
|
self.x == other.x)
|
|
|
|
def __hash__(self):
|
|
return hash((self.a, self.b))
|
|
|
|
|
|
class ObjEnum(Enum):
|
|
A = 1
|
|
B = "second"
|
|
C = "last"
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class DummyComplexClass:
|
|
prop1: str
|
|
prop2: Obj
|
|
prop3: ObjEnum
|
|
|
|
|
|
class DummyRefHelper:
|
|
"""
|
|
When something is too complicated to serialize, we just default to pickle
|
|
That is what this helper class is doing
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.refs = {}
|
|
|
|
def save_ref(self, obj):
|
|
sha256_hash = hashlib.sha256()
|
|
|
|
pickled_data = pickle.dumps(obj)
|
|
sha256_hash.update(pickled_data)
|
|
digest = sha256_hash.hexdigest()
|
|
|
|
self.refs[digest] = pickled_data
|
|
return digest
|
|
|
|
def load_ref(self, digest):
|
|
return pickle.loads(self.refs[digest])
|
|
|
|
|
|
@pytest.mark.parametrize("obj, expected", [
|
|
(1, 1),
|
|
(3.14, 3.14),
|
|
("a string", "a string"),
|
|
(True, True),
|
|
(None, None),
|
|
([1, 3.14, "a string"], [1, 3.14, "a string"]),
|
|
((1, 3.14, "a string"), {TAG_TUPLE: [1, 3.14, "a string"]}),
|
|
({1}, {TAG_SET: [1]}),
|
|
({"a": "a", "b": 3.14, "c": True}, {"a": "a", "b": 3.14, "c": True}),
|
|
({1: "a", 2: 3.14, 3: True}, {1: "a", 2: 3.14, 3: True}),
|
|
([1, [3.14, "a string"]], [1, [3.14, "a string"]]),
|
|
([1, (3.14, "a string")], [1, {TAG_TUPLE: [3.14, "a string"]}]),
|
|
([], []),
|
|
])
|
|
def test_i_can_flatten_and_restore_primitives(obj, expected):
|
|
serializer = Serializer()
|
|
|
|
flatten = serializer.serialize(obj)
|
|
assert flatten == expected
|
|
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded == obj
|
|
|
|
|
|
def test_i_can_flatten_and_restore_instances():
|
|
serializer = Serializer()
|
|
obj1 = Obj(1, "b", True)
|
|
obj2 = Obj(3.14, ("a", "b"), obj1)
|
|
|
|
flatten = serializer.serialize(obj2)
|
|
assert flatten == {TAG_OBJECT: 'tests.test_serializer.Obj',
|
|
'a': 3.14,
|
|
'b': {TAG_TUPLE: ['a', 'b']},
|
|
'c': {TAG_OBJECT: 'tests.test_serializer.Obj',
|
|
'a': 1,
|
|
'b': 'b',
|
|
'c': True}}
|
|
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded == obj2
|
|
|
|
|
|
def test_i_can_flatten_and_restore_enum():
|
|
serializer = Serializer()
|
|
obj1 = ObjEnum.A
|
|
obj2 = ObjEnum.B
|
|
obj3 = ObjEnum.C
|
|
|
|
wrapper = {
|
|
"a": obj1,
|
|
"b": obj2,
|
|
"c": obj3,
|
|
"d": obj1
|
|
}
|
|
flatten = serializer.serialize(wrapper)
|
|
assert flatten == {'a': {'__enum__': 'tests.test_serializer.ObjEnum.A'},
|
|
'b': {'__enum__': 'tests.test_serializer.ObjEnum.B'},
|
|
'c': {'__enum__': 'tests.test_serializer.ObjEnum.C'},
|
|
'd': {'__id__': 0}}
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded == wrapper
|
|
|
|
|
|
def test_i_can_flatten_and_restore_list_with_enum():
|
|
serializer = Serializer()
|
|
obj = [DummyComplexClass("a", Obj(1, "a", ObjEnum.A), ObjEnum.A),
|
|
DummyComplexClass("b", Obj(2, "b", ObjEnum.B), ObjEnum.B),
|
|
DummyComplexClass("c", Obj(3, "c", ObjEnum.C), ObjEnum.B)]
|
|
|
|
flatten = serializer.serialize(obj)
|
|
assert flatten == [{'__object__': 'tests.test_serializer.DummyComplexClass',
|
|
'prop1': 'a',
|
|
'prop2': {'__object__': 'tests.test_serializer.Obj',
|
|
'a': 1,
|
|
'b': 'a',
|
|
'c': {'__enum__': 'tests.test_serializer.ObjEnum.A'}},
|
|
'prop3': {'__id__': 2}},
|
|
{'__object__': 'tests.test_serializer.DummyComplexClass',
|
|
'prop1': 'b',
|
|
'prop2': {'__object__': 'tests.test_serializer.Obj',
|
|
'a': 2,
|
|
'b': 'b',
|
|
'c': {'__enum__': 'tests.test_serializer.ObjEnum.B'}},
|
|
'prop3': {'__id__': 5}},
|
|
{'__object__': 'tests.test_serializer.DummyComplexClass',
|
|
'prop1': 'c',
|
|
'prop2': {'__object__': 'tests.test_serializer.Obj',
|
|
'a': 3,
|
|
'b': 'c',
|
|
'c': {'__enum__': 'tests.test_serializer.ObjEnum.C'}},
|
|
'prop3': {'__id__': 5}}]
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded == obj
|
|
|
|
|
|
def test_i_can_manage_circular_reference():
|
|
serializer = Serializer()
|
|
obj1 = Obj(1, "b", True)
|
|
obj1.c = obj1
|
|
|
|
flatten = serializer.serialize(obj1)
|
|
assert flatten == {TAG_OBJECT: 'tests.test_serializer.Obj',
|
|
'a': 1,
|
|
'b': 'b',
|
|
'c': {TAG_ID: 0}}
|
|
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded.a == obj1.a
|
|
assert decoded.b == obj1.b
|
|
assert decoded.c == decoded
|
|
|
|
|
|
def test_i_can_use_refs_on_primitive():
|
|
serializer = Serializer(DummyRefHelper())
|
|
obj1 = Obj(1, "b", True)
|
|
|
|
flatten = serializer.serialize(obj1, ["c"])
|
|
assert flatten == {TAG_OBJECT: 'tests.test_serializer.Obj',
|
|
'a': 1,
|
|
'b': 'b',
|
|
'c': {TAG_REF: '112bda3b495d867b6a98c899fac7c25eb60ca4b6e6fe5ec7ab9299f93e8274bc'}}
|
|
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded == obj1
|
|
|
|
|
|
def test_i_can_use_refs_on_path():
|
|
serializer = Serializer(DummyRefHelper())
|
|
obj1 = Obj(1, "b", True)
|
|
obj2 = Obj(1, "b", obj1)
|
|
|
|
flatten = serializer.serialize(obj2, ["c.b"])
|
|
assert flatten == {TAG_OBJECT: 'tests.test_serializer.Obj',
|
|
'a': 1,
|
|
'b': 'b',
|
|
'c': {TAG_OBJECT: 'tests.test_serializer.Obj',
|
|
'a': 1,
|
|
'b': {TAG_REF: '897f2e2b559dd876ad870c82283197b8cfecdf84736192ea6fb9ee5a5080a3a4'},
|
|
'c': True}}
|
|
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded == obj2
|
|
|
|
|
|
def test_can_use_refs_when_circular_reference():
|
|
serializer = Serializer(DummyRefHelper())
|
|
obj1 = Obj(1, "b", True)
|
|
obj1.c = obj1
|
|
|
|
flatten = serializer.serialize(obj1, ["c"])
|
|
assert flatten == {TAG_OBJECT: 'tests.test_serializer.Obj',
|
|
'a': 1,
|
|
'b': 'b',
|
|
'c': {TAG_REF: "87b1980d83bd267e2c8cc2fbc435ba00349e45b736c40f3984f710ebb4495adc"}}
|
|
|
|
decoded = serializer.deserialize(flatten)
|
|
assert decoded.a == obj1.a
|
|
assert decoded.b == obj1.b
|
|
assert decoded.c == decoded
|
|
|
|
|
|
def test_i_can_serialize_date():
|
|
obj = datetime.date.today()
|
|
serializer = Serializer()
|
|
|
|
flatten = serializer.serialize(obj)
|
|
|
|
decoded = serializer.deserialize(flatten)
|
|
|
|
assert decoded == obj
|