use smart contract factory & add unit tests

multiversx_sdk_cli/cli_contracts.py

@@ -16,7 +16,7 @@
 from multiversx_sdk_cli.constants import NUMBER_OF_SHARDS
 from multiversx_sdk_cli.contract_verification import \
     trigger_contract_verification
-from multiversx_sdk_cli.contracts import SmartContract
+from multiversx_sdk_cli.contracts import SmartContract, query_contract
 from multiversx_sdk_cli.cosign_transaction import cosign_transaction
 from multiversx_sdk_cli.docker import is_docker_installed, run_docker
 from multiversx_sdk_cli.errors import DockerMissingError, NoWalletProvided
@@ -321,15 +321,6 @@ def deploy(args: Any):
     _send_or_simulate(tx, contract_address, args)
 
 
-def _prepare_contract(args: Any) -> SmartContract:
-    bytecode = utils.read_binary_file(Path(args.bytecode)).hex()
-
-    metadata = CodeMetadata(upgradeable=args.metadata_upgradeable, readable=args.metadata_readable,
-                            payable=args.metadata_payable, payable_by_sc=args.metadata_payable_by_sc)
-    contract = SmartContract(bytecode=bytecode, metadata=metadata)
-    return contract
-
-
 def _prepare_sender(args: Any) -> Account:
     sender: Account
     if args.ledger:
@@ -386,8 +377,8 @@ def call(args: Any):
     cli_shared.check_guardian_and_options_args(args)
     cli_shared.check_broadcast_args(args)
 
-    sender = _prepare_sender(args)
     cli_shared.prepare_chain_id_in_args(args)
+    sender = _prepare_sender(args)
 
     config = TransactionsFactoryConfig(args.chain)
     contract = SmartContract(config, TokenComputer())
@@ -401,35 +392,36 @@ def call(args: Any):
 
 def upgrade(args: Any):
     logger.debug("upgrade")
+    cli_shared.check_guardian_and_options_args(args)
     cli_shared.check_broadcast_args(args)
 
-    contract_address = args.contract
-    arguments = args.arguments
-    gas_price = args.gas_price
-    gas_limit = args.gas_limit
-    value = args.value
-    version = args.version
-
-    contract = _prepare_contract(args)
-    contract.address = Address.from_bech32(contract_address)
-    sender = _prepare_sender(args)
     cli_shared.prepare_chain_id_in_args(args)
+    sender = _prepare_sender(args)
 
-    tx = contract.upgrade(sender, arguments, gas_price, gas_limit, value, args.chain, version, args.guardian, args.options)
+    config = TransactionsFactoryConfig(args.chain)
+    contract = SmartContract(config, TokenComputer())
+    contract_address = Address.new_from_bech32(args.contract)
+
+    tx = contract.get_upgrade_transaction(sender, args)
     tx = _sign_guarded_tx(args, tx)
 
-    _send_or_simulate(tx, contract, args)
+    _send_or_simulate(tx, contract_address, args)
 
 
 def query(args: Any):
     logger.debug("query")
 
-    contract_address = args.contract
+    # workaround so we can use the function bellow
+    args.chain = ""
+    cli_shared.prepare_chain_id_in_args(args)
+
+    contract_address = Address.new_from_bech32(args.contract)
+
+    proxy = ProxyNetworkProvider(args.proxy)
     function = args.function
-    arguments = args.arguments
+    arguments = args.arguments or []
 
-    contract = SmartContract(address=Address.from_bech32(contract_address))
-    result = contract.query(ProxyNetworkProvider(args.proxy), function, arguments)
+    result = query_contract(contract_address, proxy, function, arguments)
     utils.dump_out_json(result)
 
 

multiversx_sdk_cli/cli_shared.py

@@ -93,9 +93,9 @@ def add_tx_args(args: List[str], sub: Any, with_nonce: bool = True, with_receive
 
 
 def add_guardian_args(sub: Any):
-    sub.add_argument("--guardian", type=str, help="the address of the guradian")
-    sub.add_argument("--guardian-service-url", type=str, help="the url of the guardian service")
-    sub.add_argument("--guardian-2fa-code", type=str, help="the 2fa code for the guardian")
+    sub.add_argument("--guardian", type=str, help="the address of the guradian", default="")
+    sub.add_argument("--guardian-service-url", type=str, help="the url of the guardian service", default="")
+    sub.add_argument("--guardian-2fa-code", type=str, help="the 2fa code for the guardian", default="")
 
 
 def add_wallet_args(args: List[str], sub: Any):

multiversx_sdk_cli/contracts.py

@@ -1,16 +1,16 @@
 import base64
 import logging
 from pathlib import Path
-from typing import Any, List, Optional, Protocol, Sequence, Tuple
+from typing import Any, List, Optional, Protocol, Sequence
 
 from multiversx_sdk_core import Transaction, TransactionPayload
 from multiversx_sdk_core.address import Address
 from multiversx_sdk_core.transaction_factories import \
     SmartContractTransactionsFactory
 from multiversx_sdk_network_providers.interface import IAddress, IContractQuery
 
-from multiversx_sdk_cli import config, errors
-from multiversx_sdk_cli.accounts import Account, EmptyAddress
+from multiversx_sdk_cli import errors
+from multiversx_sdk_cli.accounts import Account
 from multiversx_sdk_cli.constants import DEFAULT_HRP
 from multiversx_sdk_cli.utils import Object
 
@@ -88,11 +88,14 @@ def __init__(self, config: IConfig, token_computer: ITokenComputer):
         self._factory = SmartContractTransactionsFactory(config, token_computer)
 
     def get_deploy_transaction(self, owner: Account, args: Any) -> Transaction:
+        arguments = args.arguments or []
+        arguments = prepare_args_for_factory(arguments)
+
         tx = self._factory.create_transaction_for_deploy(
             sender=owner.address,
             bytecode=Path(args.bytecode),
             gas_limit=int(args.gas_limit),
-            arguments=args.arguments,
+            arguments=arguments,
             native_transfer_amount=int(args.value),
             is_upgradeable=args.metadata_upgradeable,
             is_readable=args.metadata_readable,
@@ -107,111 +110,96 @@ def get_deploy_transaction(self, owner: Account, args: Any) -> Transaction:
 
         return tx
 
-    def get_execute_transaction(self, owner: Account, args: Any) -> Transaction:
+    def get_execute_transaction(self, caller: Account, args: Any) -> Transaction:
         contract_address = Address.new_from_bech32(args.contract)
         arguments = args.arguments or []
+        arguments = prepare_args_for_factory(arguments)
 
         tx = self._factory.create_transaction_for_execute(
-            sender=owner.address,
+            sender=caller.address,
             contract=contract_address,
             function=args.function,
             gas_limit=int(args.gas_limit),
             arguments=arguments,
             native_transfer_amount=int(args.value),
             token_transfers=[]
         )
-        tx.nonce = owner.nonce
+        tx.nonce = caller.nonce
         tx.version = int(args.version)
         tx.options = int(args.options)
         tx.guardian = args.guardian
-        tx.signature = bytes.fromhex(owner.sign_transaction(tx))
+        tx.signature = bytes.fromhex(caller.sign_transaction(tx))
 
         return tx
 
-    def prepare_execute_transaction_data(self, function: str, arguments: List[Any]) -> TransactionPayload:
-        tx_data = function
-
-        for arg in arguments:
-            tx_data += f"@{_prepare_argument(arg)}"
-
-        return TransactionPayload.from_str(tx_data)
-
-    def upgrade(self, owner: Account, arguments: List[Any], gas_price: int, gas_limit: int, value: int, chain: str, version: int, guardian: str, options: int) -> Transaction:
-        self.owner = owner
-
-        arguments = arguments or []
-        gas_price = int(gas_price or config.DEFAULT_GAS_PRICE)
-        gas_limit = int(gas_limit)
-        value = value or 0
-        receiver = self.address if self.address else EmptyAddress()
-
-        tx = Transaction(
-            chain_id=chain,
-            sender=owner.address.to_bech32(),
-            receiver=receiver.to_bech32(),
-            gas_limit=gas_limit,
-            gas_price=gas_price,
-            nonce=owner.nonce,
-            amount=value,
-            data=self.prepare_upgrade_transaction_data(arguments).data,
-            version=version,
-            options=options
-        )
-
-        if guardian:
-            tx.guardian = guardian
+    def get_upgrade_transaction(self, owner: Account, args: Any):
+        contract_address = Address.new_from_bech32(args.contract)
+        arguments = args.arguments or []
+        arguments = prepare_args_for_factory(arguments)
 
+        tx = self._factory.create_transaction_for_upgrade(
+            sender=owner.address,
+            contract=contract_address,
+            bytecode=Path(args.bytecode),
+            gas_limit=int(args.gas_limit),
+            arguments=arguments,
+            native_transfer_amount=int(args.value),
+            is_upgradeable=args.metadata_upgradeable,
+            is_readable=args.metadata_readable,
+            is_payable=args.metadata_payable,
+            is_payable_by_sc=args.metadata_payable_by_sc
+        )
+        tx.nonce = owner.nonce
+        tx.version = int(args.version)
+        tx.options = int(args.options)
+        tx.guardian = args.guardian
         tx.signature = bytes.fromhex(owner.sign_transaction(tx))
-        return tx
 
-    def prepare_upgrade_transaction_data(self, arguments: List[Any]) -> TransactionPayload:
-        tx_data = f"upgradeContract@{self.bytecode}@{self.metadata.to_hex()}"
+        return tx
 
-        for arg in arguments:
-            tx_data += f"@{_prepare_argument(arg)}"
 
-        return TransactionPayload.from_str(tx_data)
+def query_contract(
+    contract_address: IAddress,
+    proxy: INetworkProvider,
+    function: str,
+    arguments: List[Any],
+    value: int = 0,
+    caller: Optional[Address] = None
+) -> List[Any]:
+    response_data = query_detailed(contract_address, proxy, function, arguments, value, caller)
+    return_data = response_data.return_data
+    return [_interpret_return_data(data) for data in return_data]
 
-    def query(
-        self,
-        proxy: INetworkProvider,
-        function: str,
-        arguments: List[Any],
-        value: int = 0,
-        caller: Optional[Address] = None
-    ) -> List[Any]:
-        response_data = self.query_detailed(proxy, function, arguments, value, caller)
-        return_data = response_data.return_data
-        return [self._interpret_return_data(data) for data in return_data]
 
-    def query_detailed(self, proxy: INetworkProvider, function: str, arguments: List[Any],
-                       value: int = 0, caller: Optional[Address] = None) -> Any:
-        arguments = arguments or []
-        # Temporary workaround, until we use sdk-core's serializer.
-        prepared_arguments = [bytes.fromhex(_prepare_argument(arg)) for arg in arguments]
+def query_detailed(contract_address: IAddress, proxy: INetworkProvider, function: str, arguments: List[Any],
+                   value: int = 0, caller: Optional[Address] = None) -> Any:
+    arguments = arguments or []
+    # Temporary workaround, until we use sdk-core's serializer.
+    prepared_arguments = [bytes.fromhex(_prepare_argument(arg)) for arg in arguments]
 
-        query = ContractQuery(self.address, function, value, prepared_arguments, caller)
+    query = ContractQuery(contract_address, function, value, prepared_arguments, caller)
 
-        response = proxy.query_contract(query)
-        # Temporary workaround, until we add "isSuccess" on the response class.
-        if response.return_code != "ok":
-            raise RuntimeError(f"Query failed: {response.return_message}")
-        return response
+    response = proxy.query_contract(query)
+    # Temporary workaround, until we add "isSuccess" on the response class.
+    if response.return_code != "ok":
+        raise RuntimeError(f"Query failed: {response.return_message}")
+    return response
 
-    def _interpret_return_data(self, data: str) -> Any:
-        if not data:
-            return data
 
-        try:
-            as_bytes = base64.b64decode(data)
-            as_hex = as_bytes.hex()
-            as_number = _interpret_as_number_if_safely(as_hex)
+def _interpret_return_data(data: str) -> Any:
+    if not data:
+        return data
 
-            result = QueryResult(data, as_hex, as_number)
-            return result
-        except Exception:
-            logger.warn(f"Cannot interpret return data: {data}")
-            return None
+    try:
+        as_bytes = base64.b64decode(data)
+        as_hex = as_bytes.hex()
+        as_number = _interpret_as_number_if_safely(as_hex)
+
+        result = QueryResult(data, as_hex, as_number)
+        return result
+    except Exception:
+        logger.warn(f"Cannot interpret return data: {data}")
+        return None
 
 
 def _interpret_as_number_if_safely(as_hex: str) -> Optional[int]:
@@ -228,6 +216,42 @@ def _interpret_as_number_if_safely(as_hex: str) -> Optional[int]:
         return None
 
 
+def prepare_execute_transaction_data(function: str, arguments: List[Any]) -> TransactionPayload:
+    tx_data = function
+
+    for arg in arguments:
+        tx_data += f"@{_prepare_argument(arg)}"
+
+    return TransactionPayload.from_str(tx_data)
+
+
+def prepare_args_for_factory(arguments: List[str]) -> List[Any]:
+    args: List[Any] = []
+
+    for arg in arguments:
+        if arg.startswith(HEX_PREFIX):
+            args.append(hex_to_bytes(arg))
+        elif arg.isnumeric():
+            args.append(int(arg))
+        elif arg.startswith(DEFAULT_HRP):
+            args.append(Address.new_from_bech32(arg))
+        elif arg.lower() == FALSE_STR_LOWER:
+            args.append(False)
+        elif arg.lower() == TRUE_STR_LOWER:
+            args.append(True)
+        elif arg.startswith(STR_PREFIX):
+            args.append(arg[len(STR_PREFIX):])
+
+    return args
+
+
+def hex_to_bytes(arg: str):
+    argument = arg[len(HEX_PREFIX):]
+    argument = argument.upper()
+    argument = ensure_even_length(argument)
+    return bytes.fromhex(argument)
+
+
 def _prepare_argument(argument: Any):
     as_str = str(argument)
     as_hex = _to_hex(as_str)
@@ -282,11 +306,3 @@ def ensure_even_length(string: str) -> str:
     if len(string) % 2 == 1:
         return '0' + string
     return string
-
-
-def sum_flag_values(flag_value_pairs: List[Tuple[int, bool]]) -> int:
-    value_sum = 0
-    for value, flag in flag_value_pairs:
-        if flag:
-            value_sum += value
-    return value_sum