contract_test.py

import sys
import time
import pprint

from web3.providers.eth_tester import EthereumTesterProvider
from web3.logs import STRICT, IGNORE, DISCARD, WARN
from web3 import Web3, HTTPProvider
from solc import compile_source
import subprocess
import json
import binascii
import threading


def _run_command(instruction):
    pipe = subprocess.Popen(instruction, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    message = pipe.stdout.read()
    error = pipe.stderr.read()
    return error, message

def deploy_contract(w3, contract_interface):
    tx_hash = w3.eth.contract(
        abi=contract_interface['abi'],
        bytecode=contract_interface['bin']).deploy()

    address = w3.eth.getTransactionReceipt(tx_hash)['contractAddress']
    return address

def web3_connection():
    web3_server = 'http://localhost:7545'
    w3 = Web3(HTTPProvider(web3_server, request_kwargs={'timeout': 240,}))
    print(w3.eth.getBlock(1))
    return w3

def compile_contract(w3):
    contract_source_path = 'contract.sol'
    error, message = _run_command('%s --evm-version byzantium --combined-json abi,bin,compact-format %s' % ('/usr/bin/solc', contract_source_path))
    compiled_sol = json.loads(message.decode('utf-8'))['contracts']

    for (function_name, contract_interface) in compiled_sol.items():
        contract = w3.eth.contract(abi=json.loads(contract_interface['abi']), bytecode=contract_interface['bin'])
        tx_hash = contract.constructor().transact(transaction={'from': Web3.toChecksumAddress(w3.eth.accounts[0]), 'gas': 4700000})
        print("Deploy Contract TX_HASH: {0}".format(binascii.hexlify(tx_hash).decode('utf-8')))
        tx_receipt = w3.eth.waitForTransactionReceipt(tx_hash)
    print("Deployed {0} to: {1}\n".format(contract_source_path, tx_receipt.contractAddress))
    compiled_contract = w3.eth.contract(
        address=tx_receipt.contractAddress,
        abi=contract_interface['abi'])
    return compiled_contract

def estimate_gas(test_var, contract):
    return contract.functions.setVar(test_var).estimateGas()

def send_transaction(w3, test_var, contract):
    tx_hash = contract.functions.setVar(test_var).transact(transaction={'from': Web3.toChecksumAddress(w3.eth.accounts[0]), 'gas': 4700000})
    receipt = w3.eth.waitForTransactionReceipt(tx_hash)
    print("var=%d => Mind in block=%s" % (test_var, receipt['blockNumber']))
    return receipt

def main():
    w3 = web3_connection()
    store_var_contract = compile_contract(w3)
    test_var = 245
    gas_estimate = estimate_gas(test_var, store_var_contract)
    if gas_estimate > 4700000:
        print("Gas cost exceeds 4700000")
        return

    print("Sending transaction to setVar(%s)\n" % (test_var))
    receipt = send_transaction(w3, test_var, store_var_contract)
    print("Was transaction successful? %s" % ('yes' if receipt['status'] else 'no'))

    # LOG TEST
    if receipt['status'] == 1:
        event_logs = getattr(store_var_contract.events, 'MyEvent')().processReceipt(receipt, errors=IGNORE)
        pprint.pprint(event_logs)
    
    event_filter = getattr(store_var_contract.events, 'MyEvent').createFilter(fromBlock=0, toBlock='latest')
    all_entries = event_filter.get_all_entries()
    pprint.pprint(all_entries)

    # THREAD TEST
    threads = []
    for test_var in range(20):
        threads.append(threading.Thread(name="[W]submit_tx", target=send_transaction, args=(w3, test_var, store_var_contract)))
        threads[test_var].start()


if __name__ == "__main__":
    main()