Added IndefiniteDecoder for round trip plutusdata serialization (#431)

* Added IndefiniteDecoder for round trip plutusdata serialization

* Fixed qa issues

* Force non-binary installation of cbor2

* Ensure pure cbor2 in CI

* Added IndefiniteList to ArrayCBORSerializable primitives

* Pass QA checks

---------

Co-authored-by: Jerry <[email protected]>

Author: theeldermillenial

.github/workflows/main.yml

@@ -28,6 +28,9 @@ jobs:
       - name: Install dependencies
         run: |
           poetry install
+      - name: Ensure pure cbor2 is installed
+        run: |
+          make ensure-pure-cbor2
       - name: Run unit tests
         run: |
           poetry run pytest --doctest-modules --ignore=examples --cov=pycardano --cov-config=.coveragerc --cov-report=xml

.github/workflows/publish.yml

@@ -24,6 +24,9 @@ jobs:
       - name: Install dependencies
         run: |
           poetry install
+      - name: Ensure pure cbor2 is installed
+        run: |
+          make ensure-pure-cbor2
       - name: Lint with flake8
         run: |
           poetry run flake8 pycardano

.gitignore

@@ -5,6 +5,7 @@ docs/build
 dist
 .mypy_cache
 coverage.xml
+.cbor2_version
 
 # IDE
 .idea

Makefile

@@ -23,10 +23,25 @@ export PRINT_HELP_PYSCRIPT
 
 BROWSER := poetry run python -c "$$BROWSER_PYSCRIPT"
 
+ensure-pure-cbor2: ## ensures cbor2 is installed with pure Python implementation
+	@poetry run python -c "from importlib.metadata import version; \
+	print(version('cbor2'))" > .cbor2_version
+	@poetry run python -c "import cbor2, inspect; \
+	print('Checking cbor2 implementation...'); \
+	decoder_path = inspect.getfile(cbor2.CBORDecoder); \
+	using_c_ext = decoder_path.endswith('.so'); \
+	print(f'Implementation path: {decoder_path}'); \
+	print(f'Using C extension: {using_c_ext}'); \
+	exit(1 if using_c_ext else 0)" || \
+	(echo "Reinstalling cbor2 with pure Python implementation..." && \
+	poetry run pip uninstall -y cbor2 && \
+	CBOR2_BUILD_C_EXTENSION=0 poetry run pip install --no-binary cbor2 "cbor2==$$(cat .cbor2_version)" --force-reinstall && \
+	rm .cbor2_version)
+
 help:
 	@python -c "$$PRINT_HELP_PYSCRIPT" < $(MAKEFILE_LIST)
 
-cov: ## check code coverage
+cov: ensure-pure-cbor2 ## check code coverage
 	poetry run pytest -n 4 --cov pycardano
 
 cov-html: cov ## check code coverage and generate an html report
@@ -54,7 +69,7 @@ clean-test: ## remove test and coverage artifacts
 	rm -fr cov_html/
 	rm -fr .pytest_cache
 
-test: ## runs tests
+test: ensure-pure-cbor2 ## runs tests
 	poetry run pytest -vv -n 4
 
 test-integration: ## runs integration tests
@@ -63,7 +78,7 @@ test-integration: ## runs integration tests
 test-single: ## runs tests with "single" markers
 	poetry run pytest -s -vv -m single
 
-qa: ## runs static analyses
+qa: ensure-pure-cbor2 ## runs static analyses
 	poetry run flake8 pycardano
 	poetry run mypy --install-types --non-interactive pycardano
 	poetry run black --check .
@@ -78,6 +93,6 @@ docs: ## build the documentation
 	poetry run sphinx-build docs/source docs/build/html
 	$(BROWSER) docs/build/html/index.html
 
-release: clean qa test format ## build dist version and release to pypi
+release: clean qa test format ensure-pure-cbor2 ## build dist version and release to pypi
 	poetry build
 	poetry publish

integration-test/run_tests.sh

@@ -6,6 +6,7 @@ set -o pipefail
 ROOT=$(pwd)
 
 poetry install -C ..
+make ensure-pure-cbor2 -f ../Makefile
 #poetry run pip install ogmios
 
 ##########

pycardano/serialization.py

@@ -21,6 +21,7 @@
     Iterable,
     List,
     Optional,
+    Sequence,
     Set,
     Type,
     TypeVar,
@@ -46,7 +47,6 @@
     CBORTag,
     FrozenDict,
     dumps,
-    loads,
     undefined,
 )
 from frozenlist import FrozenList
@@ -199,6 +199,22 @@ def wrapper(cls, value: Primitive):
 CBORBase = TypeVar("CBORBase", bound="CBORSerializable")
 
 
+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(cast(Primitive, self.decode_array(subtype=subtype)))
+    else:
+        return self.decode_array(subtype=subtype)
+
+
+try:
+    cbor2._decoder.major_decoders[4] = decode_array
+except Exception as e:
+    logger.warning("Failed to replace major decoder for indefinite array", e)
+
+
 def default_encoder(
     encoder: CBOREncoder, value: Union[CBORSerializable, IndefiniteList]
 ):
@@ -265,7 +281,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 +532,11 @@ def from_cbor(cls, payload: Union[str, bytes]) -> CBORSerializable:
         """
         if type(payload) is str:
             payload = bytes.fromhex(payload)
-        value = loads(payload)  # type: ignore
+
+        assert isinstance(payload, bytes)
+
+        value = cbor2.loads(payload)
+
         return cls.from_primitive(value)
 
     def __repr__(self):
@@ -538,7 +558,7 @@ def _restore_dataclass_field(
 
     if "object_hook" in f.metadata:
         return f.metadata["object_hook"](v)
-    return _restore_typed_primitive(f.type, v)
+    return _restore_typed_primitive(cast(Any, f.type), v)
 
 
 def _restore_typed_primitive(
@@ -580,10 +600,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)}")
@@ -712,8 +736,10 @@ def to_shallow_primitive(self) -> Primitive:
         return primitives
 
     @classmethod
-    @limit_primitive_type(list, tuple)
-    def from_primitive(cls: Type[ArrayBase], values: Union[list, tuple]) -> ArrayBase:
+    @limit_primitive_type(list, tuple, IndefiniteList)
+    def from_primitive(
+        cls: Type[ArrayBase], values: Union[list, tuple, IndefiniteList]
+    ) -> ArrayBase:
         """Restore a primitive value to its original class type.
 
         Args:

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,30 @@ class Test1(CBORSerializable):
     obj.validate()
 
 
+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):

test/pycardano/test_transaction.py

@@ -417,6 +417,23 @@ def test_multi_asset_comparison():
         a <= 1
 
 
+def test_datum_witness():
+    @dataclass
+    class TestDatum(PlutusData):
+        CONSTR_ID = 0
+        a: int
+        b: bytes
+
+    tx_body = make_transaction_body()
+    signed_tx = Transaction(
+        tx_body,
+        TransactionWitnessSet(vkey_witnesses=None, plutus_data=[TestDatum(1, b"test")]),
+    )
+    restored_tx = Transaction.from_cbor(signed_tx.to_cbor())
+
+    assert signed_tx.to_cbor_hex() == restored_tx.to_cbor_hex()
+
+
 def test_values():
     a = Value.from_primitive(
         [1, {b"1" * SCRIPT_HASH_SIZE: {b"Token1": 1, b"Token2": 2}}]