Merge pull request #147 from multiversx/TOOL-337-update-tx-on-network

Update TransactionOnNetwork to specs

multiversx_sdk/core/address.py

@@ -21,12 +21,27 @@ def __init__(self, pubkey: bytes, hrp: str) -> None:
         Args:
             pubkey (bytes): the sequence of bytes\n
             hrp (str): the human readable part"""
+
+        # used for creating an empty address
+        if not len(pubkey):
+            self.pubkey = bytes()
+            self.hrp = DEFAULT_HRP
+            return
+
         if len(pubkey) != PUBKEY_LENGTH:
             raise BadPubkeyLengthError(len(pubkey), PUBKEY_LENGTH)
 
         self.pubkey = bytes(pubkey)
         self.hrp = hrp
 
+    @classmethod
+    def empty(cls,) -> 'Address':
+        """
+        Creates an empty address object.
+        Generally speaking, this should not be used by client code **(internal use only)**.
+        """
+        return Address(b"", "")
+
     @classmethod
     def new_from_bech32(cls, value: str) -> 'Address':
         """Creates an address object from the bech32 representation of an address.

multiversx_sdk/core/address_test.py

@@ -19,11 +19,15 @@ def test_address():
     assert "erd1l453hd0gt5gzdp7czpuall8ggt2dcv5zwmfdf3sd3lguxseux2fsmsgldz" == address.to_bech32()
 
     with pytest.raises(BadPubkeyLengthError):
-        address = Address(bytes(), "erd")
+        address = Address(bytes.fromhex("fd691bb5e85d102687d8"), "erd")
 
     with pytest.raises(BadAddressError):
         address = Address.new_from_bech32("bad")
 
+    address = Address.empty()
+    assert address.pubkey == bytes()
+    assert address.hrp == "erd"
+
 
 def test_address_with_custom_hrp():
     address = Address.new_from_hex("0139472eff6886771a982f3083da5d421f24c29181e63888228dc81ca60d69e1", "test")

multiversx_sdk/core/tokens.py

@@ -61,7 +61,9 @@ def _split_identifier_into_components(self, token_parts: list[str]) -> tuple[Uni
         The second element is the ticker.
         The third element is the random sequence.
         """
-        if token_parts[0].isalnum() and token_parts[0].islower():
+        has_prefix = token_parts[0].isalnum() and token_parts[0].islower()
+
+        if has_prefix:
             return token_parts[0], token_parts[1], token_parts[2]
         return None, token_parts[0], token_parts[1]
 

multiversx_sdk/core/transaction_on_network.py

@@ -1,128 +1,60 @@
-from typing import Any, Callable, Optional
+from dataclasses import dataclass
+from typing import Any, Callable
 
 from multiversx_sdk.core.address import Address
 from multiversx_sdk.core.transaction_status import TransactionStatus
 
 
-class EmptyAddress(Address):
-    def __init__(self):
-        ...
-
-    def to_bech32(self) -> str:
-        return ""
-
-    def to_hex(self) -> str:
-        return ""
-
-    def hex(self) -> str:
-        return ""
-
-    def get_public_key(self) -> bytes:
-        return b""
-
-    def get_hrp(self) -> str:
-        return ""
-
-    def is_smart_contract(self) -> bool:
-        return False
-
-    def __eq__(self, other: object) -> bool:
-        if not isinstance(other, EmptyAddress):
-            return False
-        return True
-
-
-class TransactionOnNetwork:
-    def __init__(self) -> None:
-        self.is_completed: Optional[bool] = None
-        self.hash: str = ""
-        self.type: str = ""
-        self.nonce: int = 0
-        self.round: int = 0
-        self.epoch: int = 0
-        self.value: int = 0
-        self.receiver: Address = EmptyAddress()
-        self.sender: Address = EmptyAddress()
-        self.gas_limit: int = 0
-        self.gas_price: int = 0
-        self.data: str = ""
-        self.signature: str = ""
-        self.status: TransactionStatus = TransactionStatus("")
-        self.timestamp: int = 0
-        self.function: str = ""
-
-        self.block_nonce: int = 0
-        self.hyperblock_nonce: int = 0
-        self.hyperblock_hash: str = ""
-
-        self.contract_results: list[SmartContractResult] = []
-        self.logs: TransactionLogs = TransactionLogs()
-        self.raw_response: dict[str, Any] = {}
-
-    def get_status(self) -> TransactionStatus:
-        return self.status
-
-    def to_dictionary(self) -> dict[str, Any]:
-        return {
-            "isCompleted": self.is_completed,
-            "hash": self.hash,
-            "type": self.type,
-            "nonce": self.nonce,
-            "round": self.round,
-            "epoch": self.epoch,
-            "value": self.value,
-            "receiver": self.receiver.to_bech32(),
-            "sender": self.sender.to_bech32(),
-            "gasLimit": self.gas_limit,
-            "gasPrice": self.gas_price,
-            "data": self.data,
-            "signature": self.signature,
-            "status": self.status.status,
-            "timestamp": self.timestamp,
-            "blockNonce": self.block_nonce,
-            "hyperblockNonce": self.hyperblock_nonce,
-            "hyperblockHash": self.hyperblock_hash,
-            "smartContractResults": [item.__dict__ for item in self.contract_results],
-            "logs": self.logs.__dict__,
-        }
-
-
+@dataclass
 class TransactionEvent:
-    def __init__(self,
-                 raw: dict[str, Any] = {},
-                 address: Address = EmptyAddress(),
-                 identifier: str = "",
-                 topics: list[bytes] = [],
-                 data: bytes = b"",
-                 additional_data: list[bytes] = []) -> None:
-        self.raw = raw
-        self.address = address
-        self.identifier = identifier
-        self.topics = topics
-        self.data = data
-        self.additional_data = additional_data
+    raw: dict[str, Any]
+    address: Address
+    identifier: str
+    topics: list[bytes]
+    data: bytes
+    additional_data: list[bytes]
 
 
+@dataclass
 class TransactionLogs:
-    def __init__(self,
-                 address: Address = EmptyAddress(),
-                 events: list[TransactionEvent] = []) -> None:
-        self.address = address
-        self.events = events
+    address: Address
+    events: list[TransactionEvent]
 
 
+@dataclass
 class SmartContractResult:
-    def __init__(self,
-                 raw: dict[str, Any] = {},
-                 sender: Address = EmptyAddress(),
-                 receiver: Address = EmptyAddress(),
-                 data: bytes = b"",
-                 logs: TransactionLogs = TransactionLogs()) -> None:
-        self.raw = raw
-        self.sender = sender
-        self.receiver = receiver
-        self.data = data
-        self.logs = logs
+    raw: dict[str, Any]
+    sender: Address
+    receiver: Address
+    data: bytes
+    logs: TransactionLogs
+
+
+@dataclass
+class TransactionOnNetwork:
+    raw: dict[str, Any]
+    sender: Address
+    receiver: Address
+    hash: bytes
+    nonce: int
+    round: int
+    epoch: int
+    timestamp: int
+    block_hash: bytes
+    miniblock_hash: bytes
+    sender_shard: int
+    receiver_shard: int
+    value: int
+    gas_limit: int
+    gas_price: int
+    function: str
+    data: bytes
+    version: int
+    options: int
+    signature: bytes
+    status: TransactionStatus
+    smart_contract_results: list[SmartContractResult]
+    logs: TransactionLogs
 
 
 def find_events_by_identifier(transaction: TransactionOnNetwork, identifier: str) -> list[TransactionEvent]:
@@ -154,7 +86,7 @@ def find_events_by_predicate(
 def gather_all_events(transaction: TransactionOnNetwork) -> list[TransactionEvent]:
     all_events = [*transaction.logs.events]
 
-    for result in transaction.contract_results:
+    for result in transaction.smart_contract_results:
         all_events.extend(result.logs.events)
 
     return all_events

multiversx_sdk/core/transactions_outcome_parsers/delegation_transactions_outcome_parser_test.py

@@ -3,10 +3,11 @@
 from multiversx_sdk.core.address import Address
 from multiversx_sdk.core.transaction_on_network import (SmartContractResult,
                                                         TransactionEvent,
-                                                        TransactionLogs,
-                                                        TransactionOnNetwork)
+                                                        TransactionLogs)
 from multiversx_sdk.core.transactions_outcome_parsers.delegation_transactions_outcome_parser import \
     DelegationTransactionsOutcomeParser
+from multiversx_sdk.testutils.mock_transaction_on_network import \
+    get_empty_transaction_on_network
 from multiversx_sdk.testutils.utils import base64_topics_to_bytes
 
 
@@ -25,9 +26,12 @@ def test_parse_create_new_delegation_contract(self):
         ]
 
         delegate_event = TransactionEvent(
+            raw={},
             address=Address.new_from_bech32("erd18s6a06ktr2v6fgxv4ffhauxvptssnaqlds45qgsrucemlwc8rawq553rt2"),
             identifier="delegate",
-            topics=base64_topics_to_bytes(encodedTopics)
+            topics=base64_topics_to_bytes(encodedTopics),
+            data=b"",
+            additional_data=[]
         )
 
         encodedTopics = [
@@ -36,18 +40,24 @@ def test_parse_create_new_delegation_contract(self):
         ]
 
         sc_deploy_event = TransactionEvent(
+            raw={},
             address=Address.new_from_bech32("erd1qqqqqqqqqqqqqqqpqqqqqqqqqqqqqqqqqqqqqqqqqqqqqy8lllls62y8s5"),
             identifier="SCDeploy",
-            topics=base64_topics_to_bytes(encodedTopics)
+            topics=base64_topics_to_bytes(encodedTopics),
+            data=b"",
+            additional_data=[]
         )
 
-        logs = TransactionLogs(events=[delegate_event, sc_deploy_event])
+        logs = TransactionLogs(address=Address.empty(), events=[delegate_event, sc_deploy_event])
 
         encoded_topics = ["b2g6sUl6beG17FCUIkFwCOTGJjoJJi5SjkP2077e6xA="]
         sc_result_event = TransactionEvent(
+            raw={},
             address=Address.new_from_bech32("erd18s6a06ktr2v6fgxv4ffhauxvptssnaqlds45qgsrucemlwc8rawq553rt2"),
             identifier="completedTxEvent",
-            topics=base64_topics_to_bytes(encoded_topics)
+            topics=base64_topics_to_bytes(encoded_topics),
+            data=b"",
+            additional_data=[]
         )
 
         sc_result_log = TransactionLogs(
@@ -56,15 +66,16 @@ def test_parse_create_new_delegation_contract(self):
         )
 
         sc_result = SmartContractResult(
+            raw={},
             sender=Address.new_from_bech32("erd1qqqqqqqqqqqqqqqpqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqylllslmq6y6"),
             receiver=Address.new_from_bech32("erd18s6a06ktr2v6fgxv4ffhauxvptssnaqlds45qgsrucemlwc8rawq553rt2"),
             data=base64.b64decode(
                 "QDZmNmJAMDAwMDAwMDAwMDAwMDAwMDAwMDEwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAxMGZmZmZmZg=="),
             logs=sc_result_log
         )
 
-        tx = TransactionOnNetwork()
-        tx.contract_results = [sc_result]
+        tx = get_empty_transaction_on_network()
+        tx.smart_contract_results = [sc_result]
         tx.logs = logs
 
         outcome = self.parser.parse_create_new_delegation_contract(tx)

multiversx_sdk/core/transactions_outcome_parsers/smart_contract_transactions_outcome_parser.py

@@ -89,7 +89,7 @@ def _find_direct_sc_call_outcome_within_sc_results(
             SmartContractCallOutcome, None]:
         eligible_results: list[SmartContractResult] = []
 
-        for result in transaction.contract_results:
+        for result in transaction.smart_contract_results:
             matches_criteria_on_data = result.data.decode().startswith(ARGS_SEPARATOR)
             matches_criteria_on_receiver = result.receiver == transaction.sender
             matches_criteria_on_previous_hash = result
@@ -124,7 +124,7 @@ def _find_direct_sc_call_outcome_if_error(self, transaction: TransactionOnNetwor
         eligible_events = [event for event in transaction.logs.events if event.identifier == event_identifier]
 
         # then, we search in the logs of contract_results
-        for result in transaction.contract_results:
+        for result in transaction.smart_contract_results:
             if result.raw.get("prevTxHash", "") != transaction.hash:
                 continue
 
@@ -168,7 +168,7 @@ def _find_direct_sc_call_outcome_within_write_log_events(
         eligible_events = [event for event in transaction.logs.events if event.identifier == event_identifier]
 
         # then, we search in the logs of contract_results
-        for restult in transaction.contract_results:
+        for restult in transaction.smart_contract_results:
             if restult.raw.get("prevTxHash", "") != transaction.hash:
                 continue