Added IndefiniteDecoder for round trip plutusdata serialization

Author: theeldermillenial

pycardano/serialization.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from io import BytesIO
 import re
 import typing
 from collections import OrderedDict, UserList, defaultdict
@@ -21,6 +22,7 @@
     Iterable,
     List,
     Optional,
+    Sequence,
     Set,
     Type,
     TypeVar,
@@ -41,6 +43,7 @@
     pass
 
 from cbor2 import (
+    CBORDecoder,
     CBOREncoder,
     CBORSimpleValue,
     CBORTag,
@@ -199,6 +202,17 @@ def wrapper(cls, value: Primitive):
 CBORBase = TypeVar("CBORBase", bound="CBORSerializable")
 
 
+class IndefiniteDecoder(CBORDecoder):
+    def decode_array(self, subtype: int) -> Sequence[Any]:
+        # Major tag 4
+        length = self._decode_length(subtype, allow_indefinite=True)
+
+        if length is None:
+            return IndefiniteList(super().decode_array(subtype=subtype))
+        else:
+            return super().decode_array(subtype=subtype)
+
+
 def default_encoder(
     encoder: CBOREncoder, value: Union[CBORSerializable, IndefiniteList]
 ):
@@ -265,7 +279,7 @@ class CBORSerializable:
         does not refer to itself, which could cause infinite loops.
     """
 
-    def to_shallow_primitive(self) -> Primitive:
+    def to_shallow_primitive(self) -> Union[Primitive, CBORSerializable]:
         """
         Convert the instance to a CBOR primitive. If the primitive is a container, e.g. list, dict, the type of
         its elements could be either a Primitive or a CBORSerializable.
@@ -516,7 +530,10 @@ def from_cbor(cls, payload: Union[str, bytes]) -> CBORSerializable:
         """
         if type(payload) is str:
             payload = bytes.fromhex(payload)
-        value = loads(payload)  # type: ignore
+
+        with BytesIO(payload) as fp:
+            value = IndefiniteDecoder(fp).decode()
+
         return cls.from_primitive(value)
 
     def __repr__(self):
@@ -580,10 +597,14 @@ def _restore_typed_primitive(
             raise DeserializeException(
                 f"List types need exactly one type argument, but got {t_args}"
             )
-        t = t_args[0]
-        if not isinstance(v, list):
+        t_subtype = t_args[0]
+        if not isinstance(v, (list, IndefiniteList)):
             raise DeserializeException(f"Expected type list but got {type(v)}")
-        return IndefiniteList([_restore_typed_primitive(t, w) for w in v])
+        v_list = [_restore_typed_primitive(t_subtype, w) for w in v]
+        if t == IndefiniteList:
+            return IndefiniteList(v_list)
+        else:
+            return v_list
     elif isclass(t) and t == ByteString:
         if not isinstance(v, bytes):
             raise DeserializeException(f"Expected type bytes but got {type(v)}")

test/pycardano/test_serialization.py

@@ -30,7 +30,7 @@
     VerificationKeyWitness,
 )
 from pycardano.exception import DeserializeException, SerializeException
-from pycardano.plutus import PlutusV1Script, PlutusV2Script
+from pycardano.plutus import PlutusData, PlutusV1Script, PlutusV2Script
 from pycardano.serialization import (
     ArrayCBORSerializable,
     ByteString,
@@ -368,6 +368,31 @@ class Test1(CBORSerializable):
     obj.validate()
 
 
+@pytest.mark.xfail
+def test_datum_raw_round_trip():
+    @dataclass
+    class TestDatum(PlutusData):
+        CONSTR_ID = 0
+        a: int
+        b: List[bytes]
+
+    datum = TestDatum(a=1, b=[b"test", b"datum"])
+    restored = RawPlutusData.from_cbor(datum.to_cbor())
+    assert datum.to_cbor_hex() == restored.to_cbor_hex()
+
+
+def test_datum_round_trip():
+    @dataclass
+    class TestDatum(PlutusData):
+        CONSTR_ID = 0
+        a: int
+        b: List[bytes]
+
+    datum = TestDatum(a=1, b=[b"test", b"datum"])
+    restored = TestDatum.from_cbor(datum.to_cbor())
+    assert datum.to_cbor_hex() == restored.to_cbor_hex()
+
+
 def test_wrong_primitive_type():
     @dataclass
     class Test1(MapCBORSerializable):