source: pyson/test/ObjectMapperTest.py@ 1421

from abc import ABC
from datetime import datetime
from decimal import Decimal
import json 
import re
import sys, traceback
from typing import Dict, List, Set, Any, Union, Optional
import unittest
from uuid import uuid4, UUID

from pyson.Deserializer import Deserializer
from pyson.JsonDeserialize import JsonDeserialize
from pyson.JsonGetter import JsonGetter
from pyson.JsonSubTypes import JsonSubTypes
from pyson.JsonTypeInfo import Id, As
from pyson.JsonTypeInfo import JsonTypeInfo
from pyson.JsonValue import JsonValue
from pyson.ObjectMapper import ObjectMapper
from uri.uri import URI 


def errorsEqual( e1, e2):
    return  e1==e2 or \
        (e1.__class__==e2.__class__ and 
            e1.args == e2.args and 
            errorsEqual(e1.__cause__, e2.__cause__))
        
class Props:
    '''
    compound class with properties, used for testing 
    '''
    def __init__(self, age:int, name:str):
        if age<0:
            raise ValueError("age must be >0, got "+str(age))
        self._age=age
        self._name=name;
    def __str__(self):
        return self._name+","+str(self._age)
    def getage(self):
        return self._age
    def getname(self):
        return self._name
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._name==other._name and self._age==other._age

@JsonTypeInfo(use=Id.NAME, include=As.WRAPPER_OBJECT)
class Simple:
    def __init__(self, a:int):
        self._a=a 
    def geta(self)->int:
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._a==other._a
    def __str__(self):
        return self._name+","+str(self._a)


class SimpleWithHash(Simple):
    def __hash__(self):
        return hash(self.geta())

# define abstract root class 
# These need to be reachable globally for reference
@JsonSubTypes(["test.ObjectMapperTest.Bear"])
@JsonTypeInfo(use=Id.NAME, include=As.WRAPPER_OBJECT)
class Animal:
    pass
    
    
class Bear(Animal):
    def __init__(self, props:Props):
        self._props=props
            
    def __str__(self):
        return "Bear["+str(self._props)+"]"
        
    def getprops(self):
        return self._props
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._props==other._props



# A wrongly configured type, you must add
#@JsonSubTypes(["test.ObjectMapperTest.BadSubclass"])
class BadSuperclassMissingTypeInfo:
    pass

class BadSubclass(BadSuperclassMissingTypeInfo):
    def __init__(self, a:int):
        self._a=a
    def geta(self)->int:
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._a==other._a
    def __str__(self):
        return self._name+","+str(self._a)

#module instead of class.
@JsonSubTypes(["test.ObjectMapperTest"])
@JsonTypeInfo(use=Id.NAME, include=As.WRAPPER_OBJECT)
class BadSuperclassModuleInstead:
    pass

@JsonSubTypes(["test.ObjectMapperTest.AbstractBear"])
@JsonTypeInfo(use=Id.NAME, include=As.WRAPPER_OBJECT)
class AbstractAnimal(ABC):
    pass

class AbstractBear(AbstractAnimal):
    def __init__(self, props:Props):
        self._props=props
            
    def __str__(self):
        return "Bear["+str(self._props)+"]"
        
    def getprops(self):
        return self._props
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._props==other._props
   
    
# our parser supports non-primitive keys but python dict dont.
# therefore the following fails even before we can start testing our code...
# we need to create a hashable dict to get around ths
class mydict(dict):
    def __hash__(self, *args, **kwargs):
        return 1


# for testing @JsonValue
class Snare:
    def __init__(self, value:int):
        self._a=value
    @JsonValue()
    def getValue(self):
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and self._a==other._a

class ContainsSnare:
    def __init__(self, snare:Snare):
        self._snare=snare
    def getSnare(self):
        return self._snare
    def __eq__(self, other):
        return isinstance(other, self.__class__) and self._snare==other._snare

class OptionalVal:
    def __init__(self, value:Optional[str]): 
        self._a=value
    def getValue(self)->Optional[str]:
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and self._a==other._a


class MyDeserializer(Deserializer):
    def deserialize(self, data:object, clas: object) -> "Custom":
        if isinstance(data, float):
            return Custom(Decimal(str(data)))
        return Custom(data)


@JsonDeserialize(MyDeserializer)
class Custom:    
    def __init__(self, value:Any):
        self._a=value
    @JsonValue()
    def getValue(self):
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._a==other._a


class DefaultOne:
    '''
    Has default value
    '''
    def __init__(self, value:int=1):
        self._a=value
    def getValue(self):
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._a==other._a

class DefaultNone(ABC):
    '''
    Has default value
    '''
    def __init__(self, value:Snare=None):
        self._a=value
    def getValue(self)->Snare:
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._a==other._a

class WithUnion(ABC):
    '''
    Has default value
    '''
    def __init__(self, value:Union[Snare,None]=None):
        self._a=value
    def getValue(self)->Union[Snare,None]:
        return self._a
    def __eq__(self, other):
        return isinstance(other, self.__class__) and \
            self._a==other._a


exception=ValueError("some error")
exception.__cause__=ArithmeticError("div zero")
exceptionjson =  {"cause":{"cause":None, "message":"div zero", "stackTrace":[]},
                  "stackTrace":[],"message":"some error"}

class FrozenTest(ABC):
    def __init__(self, value:Set[str]):
        self._value=frozenset(value)
    def getValue(self) -> Set[str]:
        return self._value


class StringContainer(ABC):
    def __init__(self, val:str):
        self._val = val
    @JsonValue()
    def getValue(self) -> str:
        return self._val













class ObjectMapperTest(unittest.TestCase):
    '''
    Test a lot of back-and-forth cases.
    FIXME Can we make this a parameterized test?
    '''
    pyson=ObjectMapper()

    
    def testPrimitives(self):
        
        res=self.pyson.parse(3, int) 
        self.assertEqual(3, res)
        
    
        # this throws correct,
        self.assertRaises(ValueError, lambda:self.pyson.parse(3, str))
        
        # this throws correct,
        self.assertRaises(ValueError, lambda:self.pyson.parse("ja", int))
        
        #DEMO with nested classes of different types.
        res=self.pyson.parse('three', str)
        print(res, type(res))
        
        self.pyson.parse(3.0, float)
        self.pyson.parse(3.1, float)
        self.pyson.parse(3j, complex)
        self.pyson.parse(range(6), range)
        self.pyson.parse(True, bool)
        self.pyson.parse(False, bool)
        self.pyson.parse(b"Hello", bytes)
        self.pyson.parse(bytearray(b'\x00\x00\x00\x01'), bytearray)
        
        jsonlh = "http://localhost/"
        urilh = URI(jsonlh)
        self.assertEqual(urilh, self.pyson.parse(jsonlh, URI))
        
        #WARNING python is cheating us. equals succeeds but the type may be wrong!!
        # python says URI("hello")=="hello"
        self.assertEqual(jsonlh, self.pyson.toJson(urilh))
        self.assertEqual(str, type(self.pyson.toJson(urilh)))
        
    def testNone(self):
        self.assertEqual(None, self.pyson.toJson(None))
        self.assertEqual('null', json.dumps(None))
        self.assertEqual(None, json.loads('null'))
        self.assertEqual(None, self.pyson.parse(None, Any))
    
    def testDecimal(self):
        self.assertEqual(1.1, self.pyson.toJson(Decimal('1.1')))
        self.assertEqual(Decimal('1.1'), self.pyson.parse(1.1, Decimal))
  
        self.assertEqual(Decimal('1200'), self.pyson.parse(1200, Decimal))
        v=self.pyson.toJson(Decimal('1200'))
        self. assertEqual(1200, v)
        
        # 3/40000 triggers scientific notation in python
        v=self.pyson.toJson(Decimal(3/40000.))
        print(str(v)) # this will print "7.5e-05"
        # this notation is allowed in json file notation too.
        
    def testDecimalSerializationType(self):
        v=self.pyson.toJson(Decimal('1200'))
        self.assertEqual(int, type(v))
  
        v=self.pyson.toJson(Decimal('1200.1'))
        self.assertEqual(float, type(v))
        
        
    def testPrimitiveUUID(self):
        id=uuid4()
        self.assertEqual(id, self.pyson.parse(str(id), UUID))
        self.assertEqual(str(id), self.pyson.toJson(id))
    
    def testProps(self):
        propsjson={'age': 10, 'name': 'pietje'}
        props=Props(10, "pietje")
        self.assertEqual(propsjson,self.pyson.toJson(props))
        self.assertEqual(props, self.pyson.parse(propsjson, Props))
        
    def testParseDeepError(self):
        propsjson={'age': 10, 'name': 12}
        try:
            self.pyson.parse(propsjson, Props)
            raise AssertionError("parser did not throw")
        except ValueError as e:
            # we catch this to assure the exception contains
            # both top error and details.
            print("received error "+str(e))
            self.assertTrue(str(e).find("Error parsing"))
            self.assertTrue(str(e).find("ValueError"))
            self.assertTrue(str(e).find("expected")) 

    
    def testEmpty(self):

        class EmptyClass:
            def __init__(self):
                pass
            def __eq__(self, other):
                return isinstance(other, self.__class__)
        
        obj=EmptyClass()
        print(self.pyson.toJson(obj))
        res=self.pyson.parse({}, EmptyClass)
        self.assertEqual(obj, res)

    def testSubType(self):
       
        class Cat():
            def __init__(self, props:Props):
                self._props=props
                    
            def __str__(self):
                return "Cat["+str(self._props)+"]"
                
            def getprops(self):
                return self._props
        
        obj=Cat(Props(1,'bruno'))
        print(self.pyson.toJson(obj))
        
        
        bson={'props':{'age':1, 'name':'bruno'}}
        res=self.pyson.parse(bson, Cat)
        print(res, type(res))
        self.assertEqual(type(res.getprops()), Props)
        
    def testDeserializeNoSuchField(self):
        # Bear has a 'props' field, bot 'b'
        self.assertRaises(ValueError, lambda:self.pyson.parse({'b':1}, Props))
        #self.pyson.parse({'b':1}, Props)
                                                          
    def testInheritance(self):
        obj=Bear(Props(1,'bruno'))
        res=self.pyson.toJson(obj)
        print("result:"+str(res))
        bson={'Bear': {'props': {'age': 1, 'name': 'bruno'}}}
        self.assertEqual(bson, res)

        res=self.pyson.parse(bson, Animal)
        print("Deserialized an Animal! -->"+str(res))
        self. assertEqual(obj, res)
        
    def testAbcInheritance(self):
        obj=AbstractBear(Props(1,'bruno'))
        res=self.pyson.toJson(obj)
        print("result:"+str(res))
        bson={'AbstractBear': {'props': {'age': 1, 'name': 'bruno'}}}
        self.assertEqual(bson, res)

        res=self.pyson.parse(bson, AbstractAnimal)
        print("Deserialized an Animal! -->"+str(res))
        self. assertEqual(obj, res)
    
        
        
    def testUntypedList(self):
        class Prim:
            def __init__(self, a:list):
                self._a=a
            def geta(self)->list:
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and self._a==other._a

        obj=Prim([1,2])
        objson = {'a':[1,2]}

        self.assertEqual(objson, self.pyson.toJson(obj))
        self.assertEqual(obj, self.pyson.parse(objson, Prim))

    def testDateTime(self):
        objson = 1000120 # 1000.12ms since 1970
        obj=datetime.fromtimestamp(objson/1000.0);
        self.assertEqual(objson, self.pyson.toJson(obj))
        self.assertEqual(obj, self.pyson.parse(objson, datetime))
        
    def testDateTime2(self):
        for t in range(999, 1010):
            obj=datetime.fromtimestamp(t/1000.)
            self.assertEqual(t, self.pyson.toJson(obj))


    def testTypedList(self):
        '''
        deserializes typed list contained in another object
        '''
        class Prim:
            def __init__(self, a:List[str]):
                self._a=a
            def geta(self)->List[str]:
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a
                    
        obj=Prim(["x","y"])
        objson = {'a':["x","y"]}

        self.assertEqual(objson, self.pyson.toJson(obj))
        self.assertEqual(obj, self.pyson.parse(objson, Prim))

    def testTypedListDirect(self):
        '''
        deserializes typed list directly
        '''
                    
        obj=["x","y"]
        objson = ["x","y"]

        self.assertEqual(objson, self.pyson.toJson(obj))
        self.assertEqual(obj, self.pyson.parse(objson, List[str]))
        
    def testMixedDict(self):
        obj={'a':1, 'b':'blabla'}

        # primitive types, should not be changed
        self.assertEqual(obj, self.pyson.toJson(obj))
        self.assertEqual(obj, self.pyson.parse(obj, Dict[Any,Any]))

    def testTypedListOfObjMissingAnnotation(self):
        class Prim:
            def __init__(self, a:int):
                self._a=a
            def geta(self)->int:
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a
        obj=[Prim(1),Prim(3)]
        objson = [{"Prim":{'a':1}},{"Prim":{'a':3}}]

        # SHOULD WE CHECK THIS?        
        #self.assertRaises(ValueError, lambda:self.pyson.toJson(obj))
        # object misses annotation, therefore this will try to parse
        # Prim objects without header here.
        self.assertRaises(ValueError, lambda:self.pyson.parse(objson, List[Prim]))

    def testTypedListOfObj(self):
        @JsonTypeInfo(use=Id.NAME, include=As.WRAPPER_OBJECT)
        class Prim:
            def __init__(self, a:int):
                self._a=a
            def geta(self)->int:
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a
                    
        obj=[Prim(1),Prim(3)]
        objson = [{"Prim":{'a':1}},{"Prim":{'a':3}}]
        self.assertEqual(objson, self.pyson.toJson(obj))
        self.assertEqual(obj, self.pyson.parse(objson, List[Prim]))

    def testTypedSetOfObj(self):
        @JsonTypeInfo(use=Id.NAME, include=As.WRAPPER_OBJECT)
        class Prim:
            def __init__(self, a:int):
                self._a=a
            def geta(self)->int:
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a
                    
        obj=set([SimpleWithHash(1),SimpleWithHash(3)])
        objson = [{"SimpleWithHash":{'a':1}},{"SimpleWithHash":{'a':3}}]
        self.assertEqual(objson, self.pyson.toJson(obj))
        parsedobj=self.pyson.parse(objson, Set[SimpleWithHash])
        self.assertEqual(obj, parsedobj)


    def testExpectListButGiveDict(self):
        @JsonTypeInfo(use=Id.NAME, include=As.WRAPPER_OBJECT)
        class Prim:
            def __init__(self, a:int):
                self._a=a
            def geta(self)->int:
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a
                    
        objson = { 'a':{"Prim":{'a':1}},'c':{"Prim":{'a':3}}}
        # we request List but obj is a dict.
        self.assertRaises(ValueError,lambda:self.pyson.parse(objson, List[Prim]))

    def testSerializeDict(self):
        obj={'a':Simple(1),'c':Simple(3)}
        objson = { 'a':{"Simple":{'a':1}},'c':{"Simple":{'a':3}}}
        self.assertEqual(objson, self.pyson.toJson(obj))


    def testTypedDictOfObj(self):
        obj={'a':Simple(1),'c':Simple(3)}
        objson = { 'a':{"Simple":{'a':1}},'c':{"Simple":{'a':3}}}
        self.assertEqual(obj, self.pyson.parse(objson, Dict[str,Simple]))
        print("deserialized obj"+str(objson)+"="+str(obj))
        
    def testTypedDictSimpleKey(self):
        key=mydict()
        key["Simple"]={'a':1}
        # simple is not hashable
        objson = { key : 'a' }
        self.assertRaises(ValueError,lambda:self.pyson.parse(objson, Dict[Simple,str]))
        
    def testTypedDictSimpleKeyHashable(self):
        # key is now an object. Pyson has special provision to handle these.
        # the object is serialized to a string
        obj={SimpleWithHash(1):'a'}
        objson={"{\"SimpleWithHash\": {\"a\": 1}}": "a"}
        
        print("to obj:"+json.dumps(self.pyson.toJson(obj)))
        self.assertEqual(objson,self.pyson.toJson(obj))
        
        self.assertEqual(obj, self.pyson.parse(objson, Dict[SimpleWithHash,str]))

    def testDeserializeBadSubclass(self):
        objson= { 'BadSubclass':{ 'a':1}}
        # FIXME the error message is poor in this case. 
        self.assertRaises(ValueError,lambda:self.pyson.parse(objson, BadSuperclassMissingTypeInfo))

    
    def testModuleInsteadOfClassAsSubclasses(self):
        objson= { 'BadSubclass':{ 'a':1}}
        self.assertRaises(TypeError,lambda:self.pyson.parse(objson, BadSuperclassModuleInstead))

        

    def testJsonGetter(self):
        class Getter:
            def __init__(self, a:int):
                self._a=a
            @JsonGetter("a")
            def getValue(self):
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a
            
        getter=Getter(17)
        objson={'a':17}
        self.assertEqual(objson, self.pyson.toJson(getter))
        self.assertEqual(getter, self.pyson.parse(objson, Getter))
        
    def testGetterIgnoresCase(self):
        class Getter:
            def __init__(self, value:int):
                self._a=value
            def getValue(self):
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a
            
        getter=Getter(17)
        objson={'value':17}
        self.assertEqual(objson, self.pyson.toJson(getter))
        self.assertEqual(getter, self.pyson.parse(objson, Getter))
        
    
    def testJsonValue(self):
        getter=Snare(17)
        objson=17
        print(self.pyson.toJson(getter))
        self.assertEqual(objson, self.pyson.toJson(getter))
        val:Snare = self.pyson.parse(objson, Snare)
        self.assertEqual(getter, val)
        
                
    def testJsonValueAsPart(self):
        csnare=ContainsSnare(Snare(18))
        objson={"snare":18}
        print(self.pyson.toJson(csnare))
        self.assertEqual(objson, self.pyson.toJson(csnare))
        self.assertEqual(csnare, self.pyson.parse(objson, ContainsSnare))

            
    def testUriJsonValue(self):
        class UriJson:
            def __init__(self, value:URI):
                self._a=value
            @JsonValue()
            def getValue(self):
                return self._a
            def __eq__(self, other):
                return isinstance(other, self.__class__) and \
                    self._a==other._a

        objson="http://test/"
        obj=UriJson(URI(objson))
        self.assertEqual(objson, self.pyson.toJson(obj))
        self.assertEqual(obj, self.pyson.parse(objson, UriJson))


    def testJsonValueInList(self):
        snare1=Snare(1)
        snare2=Snare(2)
        snareList = [snare1, snare2]
        self.assertEqual([1, 2],self.pyson.toJson(snareList))
        self.assertEqual(snareList, self.pyson.parse([1,2], List[Snare]))
        


        
    def testJsonDeserializerAnnotation(self):
        self.assertEqual(Custom(Decimal(1)), self.pyson.parse(1.0, Custom))
        self.assertEqual(1.0, self.pyson.toJson(Custom(Decimal(1))))
        self.assertEqual(Custom('bla'), self.pyson.parse('bla', Custom))
        self.assertEqual('bla', self.pyson.toJson(Custom('bla')))
        
        
    def testDeserializeMissingValue(self):
        a=DefaultOne()
        self.assertEqual(1, a.getValue())
        self.assertEqual(a, self.pyson.parse({}, DefaultOne))
        
        
    def testDeserializeDefaultNone(self):
        a=DefaultNone()
        self.assertEqual(None, a.getValue())
        self.assertEqual(a, self.pyson.parse({}, DefaultNone))
        
    def testDeserializeUnion(self):
        a=WithUnion(None)
        self.assertEqual(None, a.getValue())
        self.assertEqual(a, self.pyson.parse({}, WithUnion))
    
    def testDeserializeOptional(self):
        NoneType=type(None)
        defaultone= self.pyson.parse(None, Union[DefaultOne, NoneType])
        self.assertEqual(None, defaultone)
        self.assertEqual(OptionalVal(None), self.pyson.parse({"value":None}, OptionalVal))
    
    def testSerializeOptional(self):
        val=OptionalVal(None)
        self.assertEqual({"value":None}, self.pyson.toJson(val))
        
    def testSerializeExc(self):
        self.assertEqual(exceptionjson, self.pyson.toJson(exception))
        self.assertEqual(exceptionjson, self.pyson.toJson(exception))

    def testDeserializeExc(self):
        self.assertTrue(errorsEqual(ValueError("err"),  
                self.pyson.parse({'message':'err'}, ValueError)))
        
        self.assertTrue(errorsEqual(Exception("err"), 
                self.pyson.parse({'message':'err'}, Exception)))
        self.assertEquals(None, self.pyson.parse(None, Optional[Exception]))

        e=self.pyson.parse(exceptionjson, Optional[ValueError])
        print(e)
        expected=ValueError("some error")
        expected.__cause__=Exception("div zero")

        errorsEqual(expected, e)
   
    def testFrozenSerializer(self):
        data = FrozenTest(set([1,2,3]))
        self.assertEquals({'value': [1, 2, 3]}, self.pyson.toJson(data))
        
    def testDictWithJsonValueAsKey(self):
        d:Dict[StringContainer, str]={StringContainer("test"):"value"}
        self.assertEquals({"test":"value"}, self.pyson.toJson(d))