Patch release v1.5.1.1

electrumx/lib/atomicals_blueprint_builder.py

@@ -1,4 +1,4 @@
-from typing import Optional, Tuple
+from typing import Dict, Optional, Tuple
 
 from electrumx.lib.hash import hash_to_hex_str
 from electrumx.lib.script import is_unspendable_genesis, is_unspendable_legacy
@@ -196,7 +196,9 @@ def apply_input(self, tx_in_index, sat_value, atomical_value):
         self.input_indexes.append(AtomicalInputItem(tx_in_index, sat_value, atomical_value))
 
 
-class AtomicalColoredOutputFt(IterableReprMixin):
+class AtomicalColoredOutput(IterableReprMixin):
+    type: str
+
     def __init__(
         self,
         sat_value: int,
@@ -206,7 +208,6 @@ def __init__(
         self.sat_value = sat_value
         self.atomical_value = atomical_value
         self.input_summary_info = input_summary_info
-        self.type = "FT"
 
     def __iter__(self):
         yield "type", self.type
@@ -215,17 +216,21 @@ def __iter__(self):
         yield "input_summary_info", self.input_summary_info
 
 
-class AtomicalColoredOutputNft(IterableReprMixin):
-    def __init__(self, input_summary_info: AtomicalInputSummary):
-        self.input_summary_info = input_summary_info
+class AtomicalColoredOutputFt(AtomicalColoredOutput):
+    def __init__(self, sat_value: int, atomical_value: int, input_summary_info: AtomicalInputSummary):
+        super().__init__(sat_value, atomical_value, input_summary_info)
+        self.type = "FT"
 
-    def __iter__(self):
-        yield "input_summary_info", self.input_summary_info
+
+class AtomicalColoredOutputNft(AtomicalColoredOutput):
+    def __init__(self, sat_value: int, atomical_value: int, input_summary_info: AtomicalInputSummary):
+        super().__init__(sat_value, atomical_value, input_summary_info)
+        self.type = "NFT"
 
 
 class AtomicalFtOutputBlueprintAssignmentSummary(IterableReprMixin):
     def __init__(self, outputs, fts_burned, cleanly_assigned, first_atomical_id):
-        self.outputs: dict = outputs
+        self.outputs: Dict[int, Dict[str, Dict[bytes, AtomicalColoredOutputFt]]] = outputs
         self.fts_burned: dict = fts_burned
         self.cleanly_assigned: bool = cleanly_assigned
         self.first_atomical_id: str = first_atomical_id
@@ -239,8 +244,8 @@ def __iter__(self):
 
 class AtomicalNftOutputBlueprintAssignmentSummary(IterableReprMixin):
     def __init__(self, outputs, nfts_burned=None):
-        self.outputs: dict = outputs
-        self.nfts_burned: dict = nfts_burned or {}
+        self.outputs: Dict[int, Dict[str, Dict[bytes, AtomicalColoredOutputNft]]] = outputs
+        self.nfts_burned: Dict[bytes, int] = nfts_burned or {}
 
     def __iter__(self):
         yield "outputs", self.outputs
@@ -436,9 +441,11 @@ def calculate_nft_atomicals_regular(cls, nft_map, nft_atomicals, tx, operations_
                     map_output_idxs_for_atomicals[expected_output_index] = map_output_idxs_for_atomicals.get(
                         expected_output_index
                     ) or {"atomicals": {}}
-                    map_output_idxs_for_atomicals[expected_output_index]["atomicals"][
-                        atomical_id
-                    ] = atomical_summary_info
+                    map_output_idxs_for_atomicals[expected_output_index]["atomicals"][atomical_id] = (
+                        AtomicalColoredOutputNft(
+                            atomical_summary_info.sat_value, atomical_summary_info.atomical_value, atomical_summary_info
+                        )
+                    )
                 if found_atomical_at_input:
                     next_output_idx += 1
             return AtomicalNftOutputBlueprintAssignmentSummary(map_output_idxs_for_atomicals)
@@ -454,7 +461,11 @@ def calculate_nft_atomicals_regular(cls, nft_map, nft_atomicals, tx, operations_
                 map_output_idxs_for_atomicals[expected_output_index] = map_output_idxs_for_atomicals.get(
                     expected_output_index
                 ) or {"atomicals": {}}
-                map_output_idxs_for_atomicals[expected_output_index]["atomicals"][atomical_id] = atomical_summary_info
+                map_output_idxs_for_atomicals[expected_output_index]["atomicals"][atomical_id] = (
+                    AtomicalColoredOutputNft(
+                        atomical_summary_info.sat_value, atomical_summary_info.atomical_value, atomical_summary_info
+                    )
+                )
             return AtomicalNftOutputBlueprintAssignmentSummary(map_output_idxs_for_atomicals)
 
     @classmethod
@@ -479,7 +490,9 @@ def calculate_nft_atomicals_splat(cls, nft_atomicals, tx):
             output_colored_map[expected_output_index] = output_colored_map.get(expected_output_index) or {
                 "atomicals": {}
             }
-            output_colored_map[expected_output_index]["atomicals"][atomical_id] = atomical_summary_info
+            output_colored_map[expected_output_index]["atomicals"][atomical_id] = AtomicalColoredOutputNft(
+                atomical_summary_info.sat_value, atomical_summary_info.atomical_value, atomical_summary_info
+            )
             expected_output_index_incrementing += 1
         return AtomicalNftOutputBlueprintAssignmentSummary(output_colored_map)
 
@@ -489,7 +502,7 @@ def custom_color_nft_atomicals(cls, nft_atomicals, operations_found_at_inputs, t
         output_colored_map = {}
         for atomical_id, atomical_info in sorted(nft_atomicals.items()):
             remaining_value = atomical_info.atomical_value
-            for out_idx, _txout in enumerate(tx.outputs):
+            for out_idx, tx_out in enumerate(tx.outputs):
                 compact_atomical_id = location_id_bytes_to_compact(atomical_id)
                 compact_atomical_id_data = {
                     safe_int_conversion(k, -1): safe_int_conversion(v, 0)
@@ -507,9 +520,11 @@ def custom_color_nft_atomicals(cls, nft_atomicals, operations_found_at_inputs, t
                     expected_output_index = out_idx
                 if not output_colored_map.get(expected_output_index):
                     output_colored_map[expected_output_index] = {"atomicals": {}}
-                output_colored_map[expected_output_index]["atomicals"][atomical_id] = atomical_info
+                output_colored_map[expected_output_index]["atomicals"][atomical_id] = AtomicalColoredOutputNft(
+                    tx_out.value, expected_value, atomical_info
+                )
                 remaining_value -= expected_value
-            if remaining_value > 0:
+            if remaining_value == atomical_info.atomical_value:
                 nfts_burned[atomical_id] = remaining_value
         return AtomicalNftOutputBlueprintAssignmentSummary(output_colored_map, nfts_burned)
 

electrumx/server/block_processor.py

@@ -12,7 +12,7 @@
 
 from aiorpcx import CancelledError, run_in_thread
 
-from electrumx.lib.atomicals_blueprint_builder import AtomicalsTransferBlueprintBuilder
+from electrumx.lib.atomicals_blueprint_builder import AtomicalColoredOutputNft, AtomicalsTransferBlueprintBuilder
 from electrumx.lib.hash import HASHX_LEN, double_sha256, hash_to_hex_str
 from electrumx.lib.script import (
     SCRIPTHASH_LEN,
@@ -2153,7 +2153,7 @@ def put_nft_outputs_by_blueprint(self, nft_blueprint, operations_found_at_inputs
             put_general_data(b"po" + location, txout.pk_script)
             for atomical_id, atomical_info in value_info["atomicals"].items():
                 # Only allow state or event updates if it is not immutable
-                if not atomical_info.mint_info.get("$immutable"):
+                if not atomical_info.input_summary_info.mint_info.get("$immutable"):
                     if operations_found_at_inputs:
                         if operations_found_at_inputs["op"] == "mod":
                             self.put_op_data(tx_num, tx_hash, "mod")

electrumx/server/session/session_manager.py

@@ -16,6 +16,7 @@
 
 from electrumx.lib import util
 from electrumx.lib.atomicals_blueprint_builder import (
+    AtomicalColoredOutput,
     AtomicalsTransferBlueprintBuilder,
     AtomicalsValidation,
     AtomicalsValidationError,
@@ -1125,7 +1126,10 @@ async def make_transfer_inputs(result, inputs_atomicals, tx_inputs, make_type) -
                         result[_i.txin_index].append(_data)
             return result
 
-        def make_transfer_outputs(result, outputs: Dict) -> Dict[int, List[Dict]]:
+        def make_transfer_outputs(
+            result: Dict[int, List[Dict]],
+            outputs: Dict[int, Dict[str, Dict[bytes, AtomicalColoredOutput]]],
+        ) -> Dict[int, List[Dict]]:
             for k, v in outputs.items():
                 for _atomical_id, _output in v["atomicals"].items():
                     _compact_atomical_id = location_id_bytes_to_compact(_atomical_id)

electrumx/version.py

@@ -1,4 +1,4 @@
-__version__ = "1.5.1.0"
+__version__ = "1.5.1.1"
 electrumx_version = f"ElectrumX {__version__}"
 electrumx_version_short = __version__
 

tests/lib/test_atomicals_blueprint_builder.py

@@ -1611,18 +1611,13 @@ def test_custom_colored_nft_normal():
         b"9\x12A\xaa\xbe\xa9\xe1\xccu\xd1\x86,\xc8\xc5ep)_\x92\xcdh\x8e\x0f\xeb \xc0/]\x9b\xf3\x87g\x00\x00\x00\x00"
     )
     tx, tx_hash = coin.DESERIALIZER(raw_tx, 0).read_tx_and_hash()
-    operation_found_at_inputs = parse_protocols_operations_from_witness_array(tx, tx_hash, True)
-    # try to custom nft to output 1
-    operation_found_at_inputs["payload"]["6787f39b5d2fc020eb0f8e68cd925f297065c5c82c86d175cce1a9beaa411239i0"] = {
-        1: 546
-    }
     atomicals_spent_at_inputs = {
         0: [
             {
                 "atomical_id": subject_atomical_id,
                 "location_id": b"not_used",
                 "data": b"not_used",
-                "data_value": {"sat_value": 546, "atomical_value": 546},
+                "data_value": {"sat_value": 1000, "atomical_value": 1000},
             }
         ]
     }
@@ -1634,6 +1629,11 @@ def mock_mint_fetcher(self, atomical_id):
             "type": "NFT",
         }
 
+    operation_found_at_inputs = parse_protocols_operations_from_witness_array(tx, tx_hash, True)
+    # try to custom nft to output 1
+    operation_found_at_inputs["payload"]["6787f39b5d2fc020eb0f8e68cd925f297065c5c82c86d175cce1a9beaa411239i0"] = {
+        1: 1000
+    }
     blueprint_builder = AtomicalsTransferBlueprintBuilder(
         MockLogger(),
         atomicals_spent_at_inputs,
@@ -1646,33 +1646,16 @@ def mock_mint_fetcher(self, atomical_id):
     )
     nft_output_blueprint = blueprint_builder.get_nft_output_blueprint()
     assert len(nft_output_blueprint.outputs) == 1
+    assert len(nft_output_blueprint.nfts_burned) == 0
     ft_output_blueprint = blueprint_builder.get_ft_output_blueprint()
-    assert ft_output_blueprint.cleanly_assigned == True
+    assert ft_output_blueprint.cleanly_assigned is True
     assert len(ft_output_blueprint.outputs) == 0
     assert ft_output_blueprint.fts_burned == {}
-    assert blueprint_builder.get_are_fts_burned() == False
+    assert blueprint_builder.get_are_fts_burned() is False
 
     operation_found_at_inputs["payload"]["6787f39b5d2fc020eb0f8e68cd925f297065c5c82c86d175cce1a9beaa411239i0"] = {
         "1": 546
     }
-    atomicals_spent_at_inputs = {
-        0: [
-            {
-                "atomical_id": subject_atomical_id,
-                "location_id": b"not_used",
-                "data": b"not_used",
-                "data_value": {"sat_value": 546, "atomical_value": 546},
-            }
-        ]
-    }
-
-    def mock_mint_fetcher(self, atomical_id):
-        return {
-            "atomical_id": atomical_id,
-            # set for nft
-            "type": "NFT",
-        }
-
     blueprint_builder = AtomicalsTransferBlueprintBuilder(
         MockLogger(),
         atomicals_spent_at_inputs,
@@ -1685,11 +1668,32 @@ def mock_mint_fetcher(self, atomical_id):
     )
     nft_output_blueprint = blueprint_builder.get_nft_output_blueprint()
     assert len(nft_output_blueprint.outputs) == 1
+    assert len(nft_output_blueprint.nfts_burned) == 0
     ft_output_blueprint = blueprint_builder.get_ft_output_blueprint()
-    assert ft_output_blueprint.cleanly_assigned == True
+    assert ft_output_blueprint.cleanly_assigned is True
     assert len(ft_output_blueprint.outputs) == 0
     assert ft_output_blueprint.fts_burned == {}
-    assert blueprint_builder.get_are_fts_burned() == False
+    assert blueprint_builder.get_are_fts_burned() is False
+
+    operation_found_at_inputs["payload"] = {}
+    blueprint_builder = AtomicalsTransferBlueprintBuilder(
+        MockLogger(),
+        atomicals_spent_at_inputs,
+        operation_found_at_inputs,
+        tx_hash,
+        tx,
+        mock_mint_fetcher,
+        True,
+        True,
+    )
+    nft_output_blueprint = blueprint_builder.get_nft_output_blueprint()
+    assert len(nft_output_blueprint.outputs) == 0
+    assert len(nft_output_blueprint.nfts_burned) == 1
+    ft_output_blueprint = blueprint_builder.get_ft_output_blueprint()
+    assert ft_output_blueprint.cleanly_assigned is True
+    assert len(ft_output_blueprint.outputs) == 0
+    assert ft_output_blueprint.fts_burned == {}
+    assert blueprint_builder.get_are_fts_burned() is False
 
 
 def test_partially_colored_spends_are_payments_satisfied_checks():