cic-internal-integration/apps/cic-eth/tests/fixtures_web3.py

213 lines
5.5 KiB
Python
Raw Normal View History

2021-02-01 18:12:51 +01:00
# standard imports
import os
import logging
# third-party imports
import hexbytes
import pytest
import web3
import eth_tester
from crypto_dev_signer.eth.transaction import EIP155Transaction
from crypto_dev_signer.eth.signer.defaultsigner import ReferenceSigner as EIP155Signer
from eth_keys import KeyAPI
# local imports
from cic_eth.eth import RpcClient
from cic_eth.eth.rpc import GasOracle
from cic_eth.db.models.role import AccountRole
2021-03-06 18:55:51 +01:00
from cic_eth.db.models.nonce import Nonce
2021-02-01 18:12:51 +01:00
#logg = logging.getLogger(__name__)
logg = logging.getLogger()
@pytest.fixture(scope='session')
def init_w3_nokey(
):
provider = 'http://localhost:8545'
return web3.Web3(provider)
class ProviderWalletExtension:
def __init__(self, provider, gas_price=1000000):
self.provider = provider
self.signer = EIP155Signer(provider)
self.default_gas_price = gas_price
def get(self, address, password=None):
return self.provider.get(address, password)
def new_account(self, password=None):
keys = KeyAPI()
pk = os.urandom(32)
account = self.provider.add_account(pk.hex())
self.provider.accounts[account] = keys.PrivateKey(pk)
return account
def sign_transaction(self, tx):
tx['chainId'] = int(tx['chainId'])
logg.debug('signing {}'.format(tx))
signer_tx = EIP155Transaction(tx, tx['nonce'], tx['chainId'])
tx_signed = self.signer.signTransaction(signer_tx)
tx_signed_dict = signer_tx.serialize()
tx_signed_dict['raw'] = '0x' + signer_tx.rlp_serialize().hex()
return tx_signed_dict
def sign(self, address, text=None, bytes=None):
logg.debug('sign message {} {}'.format(address[2:], text))
return self.signer.signEthereumMessage(address[2:], text)
def send_raw_transaction(self, rlp_tx_hex):
raw_tx = self.provider.backend.send_raw_transaction(bytes.fromhex(rlp_tx_hex[2:]))
return raw_tx
def gas_price(self):
return self.default_gas_price
@pytest.fixture(scope='session')
def init_wallet_extension(
init_eth_tester,
eth_provider,
):
x = ProviderWalletExtension(init_eth_tester)
def _rpcclient_web3_constructor():
w3 = web3.Web3(eth_provider)
setattr(w3.eth, 'personal', x)
setattr(w3.eth, 'sign_transaction', x.sign_transaction)
setattr(w3.eth, 'send_raw_transaction', x.send_raw_transaction)
setattr(w3.eth, 'sign', x.sign)
setattr(w3.eth, 'gas_price', x.gas_price)
return (init_eth_tester, w3)
RpcClient.set_constructor(_rpcclient_web3_constructor)
init_eth_tester.signer = EIP155Signer(x)
return x
@pytest.fixture(scope='session')
def init_w3_conn(
default_chain_spec,
init_eth_tester,
init_wallet_extension,
):
c = RpcClient(default_chain_spec)
x = ProviderWalletExtension(init_eth_tester)
# a hack to make available missing rpc calls we need
setattr(c.w3.eth, 'personal', x)
setattr(c.w3.eth, 'sign_transaction', x.sign_transaction)
setattr(c.w3.eth, 'send_raw_transaction', x.send_raw_transaction)
setattr(c.w3.eth, 'sign', x.sign)
return c.w3
@pytest.fixture(scope='function')
def init_w3(
2021-03-06 18:55:51 +01:00
init_database,
2021-02-01 18:12:51 +01:00
init_eth_tester,
init_eth_account_roles,
init_w3_conn,
):
2021-03-06 18:55:51 +01:00
for address in init_w3_conn.eth.accounts:
nonce = init_w3_conn.eth.getTransactionCount(address, 'pending')
Nonce.init(address, nonce=nonce, session=init_database)
init_database.commit()
2021-02-01 18:12:51 +01:00
yield init_w3_conn
logg.debug('mining om nom nom... {}'.format(init_eth_tester.mine_block()))
@pytest.fixture(scope='function')
def init_eth_account_roles(
init_database,
w3_account_roles,
):
2021-03-06 18:55:51 +01:00
address = w3_account_roles.get('eth_account_gas_provider')
role = AccountRole.set('GAS_GIFTER', address)
2021-02-01 18:12:51 +01:00
init_database.add(role)
2021-03-06 18:55:51 +01:00
2021-02-01 18:12:51 +01:00
return w3_account_roles
@pytest.fixture(scope='function')
def init_rpc(
default_chain_spec,
init_eth_account_roles,
init_eth_tester,
init_wallet_extension,
):
c = RpcClient(default_chain_spec)
x = ProviderWalletExtension(init_eth_tester)
# a hack to make available missing rpc calls we need
setattr(c.w3.eth, 'personal', x)
setattr(c.w3.eth, 'sign_transaction', x.sign_transaction)
setattr(c.w3.eth, 'send_raw_transaction', x.send_raw_transaction)
setattr(c.w3.eth, 'sign', x.sign)
yield c
logg.debug('mining om nom nom... {}'.format(init_eth_tester.mine_block()))
@pytest.fixture(scope='session')
def w3_account_roles(
config,
w3,
):
role_ids = [
'eth_account_bancor_deployer',
'eth_account_reserve_owner',
'eth_account_reserve_minter',
'eth_account_accounts_index_owner',
'eth_account_accounts_index_writer',
'eth_account_sarafu_owner',
'eth_account_sarafu_gifter',
'eth_account_approval_owner',
'eth_account_faucet_owner',
2021-03-06 18:55:51 +01:00
'eth_account_gas_provider',
2021-02-01 18:12:51 +01:00
]
roles = {}
i = 0
for r in role_ids:
a = w3.eth.accounts[i]
try:
a = config.get(r.upper())
except KeyError:
pass
roles[r] = a
i += 1
return roles
2021-03-06 18:55:51 +01:00
2021-02-01 18:12:51 +01:00
@pytest.fixture(scope='session')
def w3_account_token_owners(
tokens_to_deploy,
w3,
):
token_owners = {}
i = 1
for t in tokens_to_deploy:
token_owners[t[2]] = w3.eth.accounts[i]
i += 1
return token_owners