Merge pull request #9 from ElrondNetwork/legolas-addons

Add-ons for internal tests / legolas

Author: sasurobert

.gitignore

@@ -2,3 +2,12 @@
 dist
 __pycache__
 *.egg-info
+*.py[cod]
+*$py.class
+.pytest_cache/
+
+# mypy
+.mypy_cache/
+
+# Tags generated by ctags
+tags

erdpy/contracts.py

@@ -1,34 +1,35 @@
 import base64
 import logging
-from typing import Any, List
+from typing import Any, List, Optional
 
 from Cryptodome.Hash import keccak
 
 from erdpy import config, constants, errors, utils
-from erdpy.accounts import Address
+from erdpy.accounts import Account, Address
+from erdpy.interfaces import IElrondProxy
 from erdpy.transactions import Transaction
 
 logger = logging.getLogger("contracts")
 
 
 class SmartContract:
-    def __init__(self, address=None, bytecode=None, metadata=None):
+    def __init__(self, address: Optional[Address] = None, bytecode=None, metadata=None):
         self.address = Address(address)
         self.bytecode = bytecode
         self.metadata = metadata or CodeMetadata()
 
-    def deploy(self, owner, arguments, gas_price, gas_limit, value, chain, version) -> Transaction:
+    def deploy(self, owner: Account, arguments: List[Any], gas_price: int, gas_limit: int, value: int, chain: str, version: int) -> Transaction:
         self.owner = owner
         self.compute_address()
 
         arguments = arguments or []
         gas_price = int(gas_price)
         gas_limit = int(gas_limit)
-        value = str(value or "0")
+        value = value or 0
 
         tx = Transaction()
         tx.nonce = owner.nonce
-        tx.value = value
+        tx.value = str(value)
         tx.sender = owner.address.bech32()
         tx.receiver = Address.zero().bech32()
         tx.gasPrice = gas_price
@@ -40,7 +41,7 @@ def deploy(self, owner, arguments, gas_price, gas_limit, value, chain, version)
         tx.sign(owner)
         return tx
 
-    def prepare_deploy_transaction_data(self, arguments):
+    def prepare_deploy_transaction_data(self, arguments: List[Any]):
         tx_data = f"{self.bytecode}@{constants.VM_TYPE_ARWEN}@{self.metadata.to_hex()}"
 
         for arg in arguments:
@@ -59,17 +60,17 @@ def compute_address(self):
         address = bytes([0] * 8) + bytes([5, 0]) + address[10:30] + owner_bytes[30:]
         self.address = Address(address)
 
-    def execute(self, caller, function, arguments, gas_price, gas_limit, value, chain, version) -> Transaction:
+    def execute(self, caller: Account, function: str, arguments: List[str], gas_price: int, gas_limit: int, value: int, chain: str, version: int) -> Transaction:
         self.caller = caller
 
         arguments = arguments or []
         gas_price = int(gas_price)
         gas_limit = int(gas_limit)
-        value = str(value or "0")
+        value = value or 0
 
         tx = Transaction()
         tx.nonce = caller.nonce
-        tx.value = value
+        tx.value = str(value)
         tx.sender = caller.address.bech32()
         tx.receiver = self.address.bech32()
         tx.gasPrice = gas_price
@@ -81,25 +82,25 @@ def execute(self, caller, function, arguments, gas_price, gas_limit, value, chai
         tx.sign(caller)
         return tx
 
-    def prepare_execute_transaction_data(self, function, arguments):
+    def prepare_execute_transaction_data(self, function: str, arguments: List[Any]):
         tx_data = function
 
         for arg in arguments:
             tx_data += f"@{_prepare_argument(arg)}"
 
         return tx_data
 
-    def upgrade(self, owner, arguments, gas_price, gas_limit, value, chain, version) -> Transaction:
+    def upgrade(self, owner: Account, arguments: List[Any], gas_price: int, gas_limit: int, value: int, chain: str, version: int) -> Transaction:
         self.owner = owner
 
         arguments = arguments or []
         gas_price = int(gas_price or config.DEFAULT_GAS_PRICE)
         gas_limit = int(gas_limit)
-        value = str(value or "0")
+        value = value or 0
 
         tx = Transaction()
         tx.nonce = owner.nonce
-        tx.value = value
+        tx.value = str(value)
         tx.sender = owner.address.bech32()
         tx.receiver = self.address.bech32()
         tx.gasPrice = gas_price
@@ -111,30 +112,38 @@ def upgrade(self, owner, arguments, gas_price, gas_limit, value, chain, version)
         tx.sign(owner)
         return tx
 
-    def prepare_upgrade_transaction_data(self, arguments):
+    def prepare_upgrade_transaction_data(self, arguments: List[Any]):
         tx_data = f"upgradeContract@{self.bytecode}@{self.metadata.to_hex()}"
 
         for arg in arguments:
             tx_data += f"@{_prepare_argument(arg)}"
 
         return tx_data
 
-    def query(self, proxy, function, arguments) -> List[Any]:
+    def query(self, proxy: IElrondProxy, 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.get("returnData", []) or response_data.get("ReturnData", [])
+        return [self._interpret_return_data(data) for data in return_data]
+
+    def query_detailed(self, proxy: IElrondProxy, function: str, arguments: List[Any], value: int = 0, caller: Optional[Address] = None) -> Any:
         arguments = arguments or []
         prepared_arguments = [_prepare_argument(argument) for argument in arguments]
 
         payload = {
-            "ScAddress": self.address.bech32(),
-            "FuncName": function,
-            "Args": prepared_arguments
+            "scAddress": self.address.bech32(),
+            "funcName": function,
+            "args": prepared_arguments,
+            "value": str(value)
         }
 
+        if caller:
+            payload["caller"] = caller.bech32()
+
         response = proxy.query_contract(payload)
         response_data = response.get("data", {})
-        return_data = response_data.get("returnData", response_data.get("ReturnData")) or []
-        return [self._interpret_return_data(data) for data in return_data]
+        return response_data
 
-    def _interpret_return_data(self, data):
+    def _interpret_return_data(self, data: Any) -> Any:
         if not data:
             return data
 
@@ -153,7 +162,7 @@ def _interpret_return_data(self, data):
             return None
 
 
-def _prepare_argument(argument):
+def _prepare_argument(argument: Any):
     hex_prefix = "0X"
     as_string = str(argument).upper()
 
@@ -172,7 +181,7 @@ def _prepare_argument(argument):
 
 
 class CodeMetadata:
-    def __init__(self, upgradeable=True, payable=False):
+    def __init__(self, upgradeable: bool = True, payable: bool = False):
         self.upgradeable = upgradeable
         self.payable = payable
 

erdpy/interfaces.py

@@ -39,5 +39,8 @@ def send_transaction(self, payload: Any) -> str:
     def send_transactions(self, payload: List[Any]) -> Tuple[int, List[str]]:
         return 0, []
 
-    def send_transaction_and_wait_for_result(self, payload: Any, num_seconds_timeout) -> str:
+    def send_transaction_and_wait_for_result(self, payload: Any, num_seconds_timeout: int) -> str:
         return ""
+
+    def query_contract(self, payload: Any) -> Any:
+        return dict()

erdpy/projects/project_base.py

@@ -3,7 +3,7 @@
 import shutil
 from os import path
 from pathlib import Path
-from typing import List, cast
+from typing import Any, Dict, List, Union, cast
 
 from erdpy import dependencies, errors, myprocess, utils
 from erdpy.dependencies.modules import StandaloneModule
@@ -14,11 +14,11 @@
 # TODO use pathlib.Path everywhere
 class Project:
 
-    def __init__(self, directory):
+    def __init__(self, directory: Union[Path, str]):
         self.path = Path(directory).expanduser().resolve()
         self.directory = str(self.path)
 
-    def build(self, options=None):
+    def build(self, options: Union[Dict[str, Any], None] = None):
         self.options = options or dict()
         self.debug = self.options.get("debug", False)
         self._ensure_dependencies_installed()

erdpy/projects/templates.py

@@ -2,6 +2,8 @@
 import logging
 import os
 from os import path
+from pathlib import Path
+from typing import Union
 
 from erdpy import errors, utils
 from erdpy.projects import shared
@@ -33,7 +35,7 @@ def __init__(self, name, repository):
         self.language = repository.get_language(name)
 
 
-def create_from_template(name: str, template_name: str, directory: str):
+def create_from_template(name: str, template_name: str, directory: Union[Path, str]):
     directory = path.expanduser(directory)
 
     logger.info("create_from_template.name: %s", name)

erdpy/proxy/core.py

@@ -3,6 +3,7 @@
 from typing import Any, List, Tuple
 
 from erdpy.accounts import Address
+from erdpy.interfaces import IAddress
 from erdpy.proxy.http_facade import do_get, do_post
 from erdpy.proxy.messages import NetworkConfig
 
@@ -16,19 +17,19 @@ class ElrondProxy:
     def __init__(self, url: str):
         self.url = url
 
-    def get_account_nonce(self, address: Address) -> int:
+    def get_account_nonce(self, address: IAddress) -> int:
         url = f"{self.url}/address/{address.bech32()}"
         response = do_get(url)
         nonce = response.get("account").get("nonce", 0)
         return int(nonce)
 
-    def get_account_balance(self, address: Address):
+    def get_account_balance(self, address: IAddress):
         url = f"{self.url}/address/{address.bech32()}/balance"
         response = do_get(url)
         balance = response.get("balance", 0)
         return int(balance)
 
-    def get_account(self, address: Address):
+    def get_account(self, address: IAddress):
         url = f"{self.url}/address/{address.bech32()}"
         response = do_get(url)
         account = response.get("account", dict())
@@ -48,12 +49,16 @@ def get_account_transactions(self, address: Address):
 
     def get_esdt_tokens(self, address: str) -> List:
         response = do_get(f"{self.url}/address/{address}/esdt")
-        return response.get("tokens")
+        return response.get("esdts")
 
     def get_esdt_balance(self, address: str, ticker: str) -> dict:
         response = do_get(f"{self.url}/address/{address}/esdt/{ticker}")
         return response.get("tokenData")
 
+    def get_all_tokens(self) -> List[str]:
+        response = do_get(f"{self.url}/network/esdts")
+        return response.get("tokens", [])
+
     def get_num_shards(self):
         network_config = self.get_network_config()
         return network_config.num_shards
@@ -107,7 +112,7 @@ def send_transactions(self, payload: List[Any]) -> Tuple[int, List[str]]:
         hashes = response.get("txsHashes")
         return num_sent, hashes
 
-    def query_contract(self, payload: Any):
+    def query_contract(self, payload: Any) -> Any:
         url = f"{self.url}/vm-values/query"
         response = do_post(url, payload)
         return response
@@ -131,7 +136,7 @@ def get_hyperblock(self, key) -> Any:
         response = response.get("hyperblock", {})
         return response
 
-    def send_transaction_and_wait_for_result(self, payload: Any, num_seconds_timeout=100) -> str:
+    def send_transaction_and_wait_for_result(self, payload: Any, num_seconds_timeout: int = 100) -> str:
         url = f"{self.url}/transaction/send"
         response = do_post(url, payload)
         tx_hash = response.get("txHash")

erdpy/proxy/messages.py

@@ -5,5 +5,5 @@ class NetworkConfig:
     def __init__(self, data: Dict[str, Any]) -> None:
         self.num_shards = data.get("erd_num_shards_without_meta", 0)
         self.min_gas_price = data.get("erd_min_gas_price", 0)
-        self.chain_id = data.get("erd_chain_id", 0)
+        self.chain_id = data.get("erd_chain_id", "?")
         self.min_tx_version = data.get("erd_min_transaction_version", 0)

erdpy/transactions.py

@@ -2,7 +2,7 @@
 import json
 import logging
 from collections import OrderedDict
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from erdpy import errors, utils
 from erdpy.accounts import Account, Address
@@ -171,7 +171,7 @@ def wrap_inner(self, inner: ITransaction) -> None:
 
 class BunchOfTransactions:
     def __init__(self):
-        self.transactions = []
+        self.transactions: List[Transaction] = []
 
     def add_prepared(self, transaction: Transaction):
         self.transactions.append(transaction)

erdpy/utils.py

@@ -9,7 +9,7 @@
 import tarfile
 import zipfile
 from pathlib import Path
-from typing import Any, Dict, List, Union
+from typing import Any, List, Union, Optional
 
 import toml
 
@@ -75,7 +75,7 @@ def read_lines(file: str):
     return lines
 
 
-def read_file(f: Any, binary=False) -> Union[str, bytes]:
+def read_file(f: Any, binary: bool = False) -> Union[str, bytes]:
     try:
         mode = "rb" if binary else "r"
         if isinstance(f, str) or isinstance(f, pathlib.PosixPath):
@@ -102,7 +102,7 @@ def write_toml_file(filename, data):
         toml.dump(data, f)
 
 
-def read_json_file(filename: str) -> Dict[str, Any]:
+def read_json_file(filename: Union[str, Path]) -> Any:
     with open(filename) as f:
         return json.load(f)
 
@@ -146,7 +146,7 @@ def find_in_dictionary(dictionary, compound_path):
     return node
 
 
-def list_files(folder: str, suffix: str = None) -> List[str]:
+def list_files(folder: Union[str, Path], suffix: Optional[str] = None) -> List[str]:
     files = os.listdir(folder)
     files = [os.path.join(folder, f) for f in files]
 
@@ -156,7 +156,7 @@ def list_files(folder: str, suffix: str = None) -> List[str]:
     return files
 
 
-def remove_folder(folder):
+def remove_folder(folder: Union[str, Path]):
     shutil.rmtree(folder, ignore_errors=True)