import dataclasses import datetime import hashlib import pickle from enum import Enum import pandas as pd import pytest from core.serializer import TAG_TUPLE, TAG_SET, Serializer, TAG_OBJECT, TAG_ID, TAG_REF from core.settings_objects import BudgetTrackerFile, BudgetTrackerFiles 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_manage_implicit_use_refs(): data = {'Key': ['A'], 'Value': [0.1]} obj = BudgetTrackerFile(2024, 8, data=pd.DataFrame(data)) serializer = Serializer(DummyRefHelper()) flatten = serializer.serialize(obj) # use_refs is not indicated. It will be found browsing the objs assert flatten == {TAG_OBJECT: 'core.settings_objects.BudgetTrackerFile', 'month': 8, 'year': 2024, 'file_name': None, 'grid_settings': None, 'sheet_name': None, 'data': {TAG_REF: "0d523d048ce02198a511c8e647103d89f41da23dcf90127a5be7d62097f64079"}} def test_i_can_manage_implicit_use_refs_in_sub_objects(): data1 = {'Key': ['A'], 'Value': [0.1]} sub_obj1 = BudgetTrackerFile(2024, 8, data=pd.DataFrame(data1)) data2 = {'Key': ['B'], 'Value': [0.2]} sub_obj2 = BudgetTrackerFile(2024, 8, data=pd.DataFrame(data2)) obj = BudgetTrackerFiles([sub_obj1, sub_obj2]) serializer = Serializer(DummyRefHelper()) flatten = serializer.serialize(obj) # use_refs is not indicated. It will be found browsing the objs assert flatten == {TAG_OBJECT: 'core.settings_objects.BudgetTrackerFiles', 'files': [{TAG_OBJECT: 'core.settings_objects.BudgetTrackerFile', 'data': { TAG_REF: '0d523d048ce02198a511c8e647103d89f41da23dcf90127a5be7d62097f64079'}, 'month': 8, 'year': 2024, 'file_name': None, 'grid_settings': None, 'sheet_name': None, }, {TAG_OBJECT: 'core.settings_objects.BudgetTrackerFile', 'data': { TAG_REF: '51fa88e6f200da3860693e7f7765e444c55f5d4a3b6b073d9b1a9bddd742247e'}, 'month': 8, 'year': 2024, 'file_name': None, 'grid_settings': None, 'sheet_name': None, }]} def test_i_can_serialize_date(): obj = datetime.date.today() serializer = Serializer() flatten = serializer.serialize(obj) decoded = serializer.deserialize(flatten) assert decoded == obj # def test_i_can_manage_sub_class(): # obj2 = Obj2("a", "b", "x") # serializer = Serializer() # # flatten = serializer.serialize(obj2) # # decoded = serializer.deserialize(flatten) # # assert decoded == obj2