d1543ce678
* types tests * cleanup * cleanup * fix differences in the rpc calls with geth * cleanup * fix linter. getProof future queries now behave as geth * fix spelling Co-authored-by: Federico Kunze Küllmer <31522760+fedekunze@users.noreply.github.com>
98 lines
3.0 KiB
Python
98 lines
3.0 KiB
Python
import json
|
|
import os
|
|
import socket
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from eth_account import Account
|
|
from web3._utils.transactions import fill_nonce, fill_transaction_defaults
|
|
|
|
load_dotenv(Path(__file__).parent.parent.parent / "scripts/.env")
|
|
Account.enable_unaudited_hdwallet_features()
|
|
ACCOUNTS = {
|
|
"validator": Account.from_mnemonic(os.getenv("VALIDATOR1_MNEMONIC")),
|
|
"community": Account.from_mnemonic(os.getenv("COMMUNITY_MNEMONIC")),
|
|
"signer1": Account.from_mnemonic(os.getenv("SIGNER1_MNEMONIC")),
|
|
"signer2": Account.from_mnemonic(os.getenv("SIGNER2_MNEMONIC")),
|
|
}
|
|
KEYS = {name: account.key for name, account in ACCOUNTS.items()}
|
|
ADDRS = {name: account.address for name, account in ACCOUNTS.items()}
|
|
ETHERMINT_ADDRESS_PREFIX = "ethm"
|
|
TEST_CONTRACTS = {
|
|
"TestERC20A": "TestERC20A.sol",
|
|
}
|
|
|
|
|
|
def contract_path(name, filename):
|
|
return (
|
|
Path(__file__).parent
|
|
/ "contracts/artifacts/contracts/"
|
|
/ filename
|
|
/ (name + ".json")
|
|
)
|
|
|
|
|
|
CONTRACTS = {
|
|
**{
|
|
name: contract_path(name, filename) for name, filename in TEST_CONTRACTS.items()
|
|
},
|
|
}
|
|
|
|
|
|
def wait_for_port(port, host="127.0.0.1", timeout=40.0):
|
|
start_time = time.perf_counter()
|
|
while True:
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout):
|
|
break
|
|
except OSError as ex:
|
|
time.sleep(0.1)
|
|
if time.perf_counter() - start_time >= timeout:
|
|
raise TimeoutError(
|
|
"Waited too long for the port {} on host {} to start accepting "
|
|
"connections.".format(port, host)
|
|
) from ex
|
|
|
|
|
|
def w3_wait_for_new_blocks(w3, n):
|
|
begin_height = w3.eth.block_number
|
|
while True:
|
|
time.sleep(0.5)
|
|
cur_height = w3.eth.block_number
|
|
if cur_height - begin_height >= n:
|
|
break
|
|
|
|
|
|
def deploy_contract(w3, jsonfile, args=(), key=KEYS["validator"]):
|
|
"""
|
|
deploy contract and return the deployed contract instance
|
|
"""
|
|
acct = Account.from_key(key)
|
|
info = json.loads(jsonfile.read_text())
|
|
contract = w3.eth.contract(abi=info["abi"], bytecode=info["bytecode"])
|
|
tx = contract.constructor(*args).buildTransaction({"from": acct.address})
|
|
txreceipt = send_transaction(w3, tx, key)
|
|
assert txreceipt.status == 1
|
|
address = txreceipt.contractAddress
|
|
return w3.eth.contract(address=address, abi=info["abi"])
|
|
|
|
|
|
def fill_defaults(w3, tx):
|
|
return fill_nonce(w3, fill_transaction_defaults(w3, tx))
|
|
|
|
|
|
def sign_transaction(w3, tx, key=KEYS["validator"]):
|
|
"fill default fields and sign"
|
|
acct = Account.from_key(key)
|
|
tx["from"] = acct.address
|
|
tx = fill_transaction_defaults(w3, tx)
|
|
tx = fill_nonce(w3, tx)
|
|
return acct.sign_transaction(tx)
|
|
|
|
|
|
def send_transaction(w3, tx, key=KEYS["validator"]):
|
|
signed = sign_transaction(w3, tx, key)
|
|
txhash = w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
return w3.eth.wait_for_transaction_receipt(txhash)
|