Abi / codecs (integration sketch)

.github/workflows/test.yml

@@ -0,0 +1,32 @@
+name: Run tests
+
+on:
+  pull_request:
+  workflow_dispatch:
+
+jobs:
+  run-tests:
+    name: Run tests on ${{ matrix.os }}, python ${{ matrix.python-version }}
+
+    runs-on: ${{ matrix.os }}
+
+    strategy:
+      matrix:
+        os: [ubuntu-latest]
+        python-version: [3.8]
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.x'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+        pip install -r requirements-dev.txt
+    - name: Test with pytest
+      run: |
+        export PYTHONPATH=.
+        pytest .

README.md

@@ -23,6 +23,7 @@ Create a virtual environment and install the dependencies:
 python3 -m venv ./.venv
 source ./.venv/bin/activate
 pip install -r ./requirements.txt --upgrade
+pip install -r ./requirements-dev.txt --upgrade
 ```
 
 ### Operation

contracts/contract_identities.py

@@ -1,5 +1,7 @@
 from abc import abstractmethod, ABC
 from enum import Enum
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
 
 from utils.utils_chain import Account
 from multiversx_sdk import ProxyNetworkProvider
@@ -15,20 +17,20 @@ class DEXContractInterface(ABC):
     address: str = NotImplemented
 
     @abstractmethod
-    def get_config_dict(self) -> dict:
+    def get_config_dict(self) -> Dict[str, Any]:
         pass
 
     @classmethod
     @abstractmethod
-    def load_config_dict(cls, config_dict: dict):
+    def load_config_dict(cls, config_dict: Dict[str, Any]):
         pass
 
     @abstractmethod
-    def contract_deploy(self, deployer: Account, proxy: ProxyNetworkProvider, bytecode_path, args: list):
+    def contract_deploy(self, deployer: Account, proxy: ProxyNetworkProvider, bytecode_path: str | Path, args: List[Any]) -> Tuple[str, str]:
         pass
 
     @abstractmethod
-    def contract_start(self, deployer: Account, proxy: ProxyNetworkProvider, args: list = None):
+    def contract_start(self, deployer: Account, proxy: ProxyNetworkProvider, args: Optional[List[Any]] = None):
         pass
 
     @abstractmethod
@@ -149,5 +151,3 @@ class LockedAssetContractIdentity(DEXContractIdentityInterface):
     address: str
     unlocked_asset: str
     locked_asset: str
-
-

contracts/proxy_deployer_contract.py

@@ -3,7 +3,7 @@
 
 from contracts.contract_identities import DEXContractInterface
 from utils.logger import get_logger
-from utils.utils_tx import deploy, endpoint_call, get_deployed_address_from_tx
+from utils.utils_tx import deploy, endpoint_call, get_deployed_address_given_deployer
 from utils.utils_generic import log_step_fail, log_step_pass, log_warning, log_unexpected_args
 from utils.utils_chain import Account, WrapperAddress as Address
 from multiversx_sdk import CodeMetadata, ProxyNetworkProvider
@@ -79,7 +79,7 @@ def farm_contract_deploy(self, deployer: Account, proxy: ProxyNetworkProvider, a
 
         # retrieve deployed contract address
         if tx_hash != "":
-            address = get_deployed_address_from_tx(tx_hash, proxy)
+            address = get_deployed_address_given_deployer(deployer)
 
         return tx_hash, address
 

contracts/router_contract.py

@@ -1,7 +1,7 @@
 import config
 from contracts.contract_identities import DEXContractInterface, RouterContractVersion
 from utils.logger import get_logger
-from utils.utils_tx import deploy, upgrade_call, get_deployed_address_from_tx, endpoint_call
+from utils.utils_tx import deploy, upgrade_call, get_deployed_address_given_deployer, endpoint_call
 from utils.utils_generic import log_step_pass, log_unexpected_args
 from utils.utils_chain import Account, WrapperAddress as Address
 from multiversx_sdk import CodeMetadata, ProxyNetworkProvider
@@ -130,7 +130,7 @@ def pair_contract_deploy(self, deployer: Account, proxy: ProxyNetworkProvider, a
 
         # retrieve deployed contract address
         if tx_hash != "":
-            address = get_deployed_address_from_tx(tx_hash, proxy)
+            address = get_deployed_address_given_deployer(deployer)
 
         return tx_hash, address
 

contracts/simple_lock_energy_contract.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 from typing import Dict, List, Any
 
 from contracts.contract_identities import DEXContractInterface
@@ -22,7 +23,7 @@ def __init__(self, base_token: str, locked_token: str = "", lp_proxy_token: str
         self.lp_proxy_token = lp_proxy_token
         self.farm_proxy_token = farm_proxy_token
 
-    def get_config_dict(self) -> dict:
+    def get_config_dict(self) -> Dict[str, Any]:
         output_dict = {
             "address": self.address,
             "base_token": self.base_token,
@@ -33,14 +34,14 @@ def get_config_dict(self) -> dict:
         return output_dict
 
     @classmethod
-    def load_config_dict(cls, config_dict: dict):
+    def load_config_dict(cls, config_dict: Dict[str, Any]):
         return SimpleLockEnergyContract(address=config_dict['address'],
                                         base_token=config_dict['base_token'],
                                         locked_token=config_dict['locked_token'],
                                         lp_proxy_token=config_dict['lp_proxy_token'],
                                         farm_proxy_token=config_dict['farm_proxy_token'])
 
-    def contract_deploy(self, deployer: Account, proxy: ProxyNetworkProvider, bytecode_path, args: list):
+    def contract_deploy(self, deployer: Account, proxy: ProxyNetworkProvider, bytecode_path: str | Path, args: List[Any]):
         """Expecting as args:
                     type[str]: legacy token id
                     type[str]: locked asset factory address
@@ -308,7 +309,7 @@ def remove_sc_from_token_transfer_whitelist(self, deployer: Account, proxy: Prox
 
         return endpoint_call(proxy, gas_limit, deployer, Address(self.address),
                              "removeFromTokenTransferWhitelist", sc_args)
-    
+
     def set_energy_for_old_tokens(self, deployer: Account, proxy: ProxyNetworkProvider, args: list):
         """ Expected as args:
             type[str]: address
@@ -328,7 +329,7 @@ def set_energy_for_old_tokens(self, deployer: Account, proxy: ProxyNetworkProvid
             args[1],
             args[2]
         ]
-        
+
         return endpoint_call(proxy, gas_limit, deployer, Address(self.address), "setEnergyForOldTokens", sc_args)
 
     def lock_tokens(self, user: Account, proxy: ProxyNetworkProvider, args: list):

contracts/simple_lock_energy_contract_test.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+
+from contracts.simple_lock_energy_contract import SimpleLockEnergyContract
+from testutils.mock_network_provider import MockNetworkProvider
+from utils.utils_chain import Account
+
+testdata_folder = Path(__file__).parent.parent / "testdata"
+
+
+def test_deploy_contract():
+    account = Account(pem_file=testdata_folder / "alice.pem")
+    bytecode_path = testdata_folder / "dummy.wasm"
+    bytecode = bytecode_path.read_bytes()
+    network_provider = MockNetworkProvider()
+
+    contract = SimpleLockEnergyContract(
+        base_token="TEST-987654"
+    )
+
+    tx_hash, contract_address = contract.contract_deploy(
+        proxy=network_provider,
+        deployer=account,
+        bytecode_path=bytecode_path,
+        args=[
+            "TEST-123456",
+            "erd1qqqqqqqqqqqqqpgqaxa53w6uk43n6dhyt2la6cd5lyv32qn4396qfsqlnk",
+            42,
+            [360, 720, 1440],
+            [5000, 7000, 8000]
+        ]
+    )
+
+    assert tx_hash == "cbde33c54afde0a215961568755167c60255a95c70f1a8d91f0b29dc0baa37c2"
+    assert contract_address == "erd1qqqqqqqqqqqqqpgqak8zt22wl2ph4tswtyc39namqx6ysa2sd8ss4xmlj3"
+
+    tx_on_network = network_provider.get_transaction(tx_hash)
+    assert tx_on_network.data == f"{bytecode.hex()}@0500@0504@544553542d393837363534@544553542d313233343536@00000000000000000500e9bb48bb5cb5633d36e45abfdd61b4f9191502758974@2a@0168@1388@02d0@1b58@05a0@1f40"

pyrightconfig.json

@@ -0,0 +1,14 @@
+{
+  "include": ["contracts", "deploy", "events", "utils", "testutils"],
+  "exclude": ["**/__pycache__"],
+  "ignore": [],
+  "defineConstant": {
+    "DEBUG": true
+  },
+  "venvPath": ".",
+  "venv": ".venv",
+  "stubPath": "",
+  "reportMissingImports": true,
+  "reportMissingTypeStubs": false,
+  "reportUnknownParameterType": true
+}

requirements-dev.txt

@@ -0,0 +1 @@
+pytest

requirements.txt

@@ -1,4 +1,4 @@
 ipykernel
 toml
 debugpy
-multiversx-sdk @ git+https://github.com/multiversx/mx-sdk-py-incubator@main
+multiversx-sdk @ git+https://github.com/multiversx/mx-sdk-py-incubator@codecs-init

testdata/alice.pem

@@ -0,0 +1,5 @@
+-----BEGIN PRIVATE KEY for erd1qyu5wthldzr8wx5c9ucg8kjagg0jfs53s8nr3zpz3hypefsdd8ssycr6th-----
+NDEzZjQyNTc1ZjdmMjZmYWQzMzE3YTc3ODc3MTIxMmZkYjgwMjQ1ODUwOTgxZTQ4
+YjU4YTRmMjVlMzQ0ZThmOTAxMzk0NzJlZmY2ODg2NzcxYTk4MmYzMDgzZGE1ZDQy
+MWYyNGMyOTE4MWU2Mzg4ODIyOGRjODFjYTYwZDY5ZTE=
+-----END PRIVATE KEY for erd1qyu5wthldzr8wx5c9ucg8kjagg0jfs53s8nr3zpz3hypefsdd8ssycr6th-----

testutils/mock_network_provider.py

@@ -0,0 +1,62 @@
+from typing import Callable, Dict, Optional
+
+from multiversx_sdk import ProxyNetworkProvider, TransactionComputer
+from multiversx_sdk.core.address import Address
+from multiversx_sdk.network_providers.network_config import NetworkConfig
+from multiversx_sdk.network_providers.transaction_status import \
+    TransactionStatus
+from multiversx_sdk.network_providers.transactions import (
+    ITransaction, TransactionOnNetwork)
+
+
+class MockNetworkProvider(ProxyNetworkProvider):
+    def __init__(self) -> None:
+        super().__init__("https://example.multiversx.com")
+
+        self.transactions: Dict[str, TransactionOnNetwork] = {}
+        self.transaction_computer = TransactionComputer()
+
+    def get_network_config(self) -> NetworkConfig:
+        network_config = NetworkConfig()
+        network_config.chain_id = "T"
+        network_config.gas_per_data_byte = 1500
+        network_config.min_gas_limit = 50000
+        network_config.min_gas_price = 1000000000
+        return network_config
+
+    def mock_update_transaction(self, hash: str, mutate: Callable[[TransactionOnNetwork], None]) -> None:
+        transaction = self.transactions.get(hash, None)
+
+        if transaction:
+            mutate(transaction)
+
+    def mock_put_transaction(self, hash: str, transaction: TransactionOnNetwork) -> None:
+        self.transactions[hash] = transaction
+
+    def get_transaction(self, tx_hash: str, with_process_status: Optional[bool] = False) -> TransactionOnNetwork:
+        transaction = self.transactions.get(tx_hash, None)
+        if transaction:
+            return transaction
+
+        raise Exception("Transaction not found")
+
+    def get_transaction_status(self, tx_hash: str) -> TransactionStatus:
+        transaction = self.get_transaction(tx_hash)
+        return transaction.status
+
+    def send_transaction(self, transaction: ITransaction) -> str:
+        hash = self.transaction_computer.compute_transaction_hash(transaction).hex()
+
+        transaction_on_network = TransactionOnNetwork()
+        transaction_on_network.hash = hash
+        transaction_on_network.sender = Address.from_bech32(transaction.sender)
+        transaction_on_network.receiver = Address.from_bech32(transaction.receiver)
+        transaction_on_network.value = transaction.value
+        transaction_on_network.gas_limit = transaction.gas_limit
+        transaction_on_network.gas_price = transaction.gas_price
+        transaction_on_network.data = transaction.data.decode("utf-8")
+        transaction_on_network.signature = transaction.signature.hex()
+
+        self.mock_put_transaction(hash, transaction_on_network)
+
+        return hash

utils/utils_chain.py

@@ -38,9 +38,9 @@ def __repr__(self):
 class Account:
     def __init__(self,
                  address: Optional[str] = None,
-                 pem_file: Optional[str] = None,
+                 pem_file: Optional[str | Path] = None,
                  pem_index: int = 0,
-                 key_file: str = "",
+                 key_file: str | Path = "",
                  password: str = "",
                  ledger: bool = False):
         self.address = Address.new_from_bech32(address) if address else None

utils/utils_tx.py

@@ -2,10 +2,11 @@
 import time
 import traceback
 from pathlib import Path
-from typing import Dict, List, Tuple, Union
+from typing import Any, Dict, List, Tuple, Union
 
-from multiversx_sdk import (Address, ApiNetworkProvider, GenericError,
-                            ProxyNetworkProvider, TokenPayment, Transaction)
+from multiversx_sdk import (Address, AddressComputer, ApiNetworkProvider,
+                            GenericError, ProxyNetworkProvider, TokenPayment,
+                            Transaction)
 from multiversx_sdk.core.interfaces import ICodeMetadata
 from multiversx_sdk.core.transaction_builders import (
     ContractCallBuilder, ContractDeploymentBuilder, ContractUpgradeBuilder,
@@ -471,16 +472,16 @@ def endpoint_call(proxy: ProxyNetworkProvider, gas: int, user: Account, contract
 
 
 def deploy(contract_label: str, proxy: ProxyNetworkProvider, gas: int,
-           owner: Account, bytecode_path: str, metadata: ICodeMetadata, args: list) -> Tuple[str, str]:
+           owner: Account, bytecode_path: str | Path, metadata: ICodeMetadata, args: List[Any]) -> Tuple[str, str]:
     logger.debug(f"Deploy {contract_label}")
     network_config = proxy.get_network_config()     # TODO: find solution to avoid this call
     tx_hash, contract_address = "", ""
 
     tx = prepare_deploy_tx(owner, network_config, gas, Path(bytecode_path), metadata, args)
     tx_hash = send_deploy_tx(tx, proxy)
+    contract_address = get_deployed_address_given_deployer(owner)
 
     if tx_hash:
-        contract_address = get_deployed_address_from_tx(tx_hash, proxy)
         owner.nonce += 1
 
     return tx_hash, contract_address
@@ -541,11 +542,12 @@ def get_event_from_tx(event_id: str, tx_hash: str, proxy: ProxyNetworkProvider)
     return event
 
 
-def get_deployed_address_from_tx(tx_hash: str, proxy: ProxyNetworkProvider) -> str:
-    event = get_event_from_tx("SCDeploy", tx_hash, proxy)
-    if event is None:
-        return ""
-    return event.address.to_bech32()
+
+def get_deployed_address_given_deployer(deployer: Account) -> str:
+    address_computer = AddressComputer()
+    assert deployer.address is not None
+    contract_address = address_computer.compute_contract_address(deployer.address, deployer.nonce).to_bech32()
+    return contract_address
 
 
 def broadcast_transactions(transactions: List[Transaction], proxy: ProxyNetworkProvider,