cic-stack/apps/cic-eth/tests/tasks/test_account.py
2021-03-06 17:55:51 +00:00

139 lines
3.2 KiB
Python

# standard imports
import os
import logging
import time
# third-party imports
import pytest
import web3
import celery
# local imports
from cic_eth.error import OutOfGasError
from cic_eth.db.models.otx import Otx
from cic_eth.db.models.base import SessionBase
from cic_eth.db.enum import StatusEnum
from cic_eth.db.enum import StatusEnum
from cic_eth.db.models.nonce import Nonce
from cic_eth.db.models.role import AccountRole
from cic_eth.eth.account import AccountTxFactory
logg = logging.getLogger() #__name__)
def test_create_account(
default_chain_spec,
init_w3,
init_database,
celery_session_worker,
):
s = celery.signature(
'cic_eth.eth.account.create',
[
'foo',
str(default_chain_spec),
],
)
t = s.apply_async()
r = t.get()
logg.debug('got account {}'.format(r))
session = SessionBase.create_session()
q = session.query(Nonce).filter(Nonce.address_hex==r)
o = q.first()
logg.debug('oooo s {}'.format(o))
session.close()
assert o != None
assert o.nonce == 0
s = celery.signature(
'cic_eth.eth.account.have',
[
r,
str(default_chain_spec),
],
)
t = s.apply_async()
assert r == t.get()
def test_register_account(
default_chain_spec,
accounts_registry,
init_database,
init_eth_tester,
init_w3,
init_rpc,
cic_registry,
celery_session_worker,
eth_empty_accounts,
):
logg.debug('chainspec {}'.format(str(default_chain_spec)))
s_nonce = celery.signature(
'cic_eth.eth.tx.reserve_nonce',
[
eth_empty_accounts[0],
init_w3.eth.accounts[0],
],
queue=None,
)
s_register = celery.signature(
'cic_eth.eth.account.register',
[
str(default_chain_spec),
init_w3.eth.accounts[0],
],
)
s_nonce.link(s_register)
t = s_nonce.apply_async()
address = t.get()
for r in t.collect():
pass
assert t.successful()
session = SessionBase.create_session()
o = session.query(Otx).first()
tx_signed_hex = o.signed_tx
session.close()
s_send = celery.signature(
'cic_eth.eth.tx.send',
[
[tx_signed_hex],
str(default_chain_spec),
],
)
t = s_send.apply_async()
address = t.get()
r = t.collect()
t.successful()
init_eth_tester.mine_block()
assert accounts_registry.have(eth_empty_accounts[0])
def test_role_task(
init_database,
celery_session_worker,
default_chain_spec,
):
address = '0x' + os.urandom(20).hex()
role = AccountRole.set('foo', address)
init_database.add(role)
init_database.commit()
s = celery.signature(
'cic_eth.eth.account.role',
[
address,
str(default_chain_spec),
],
)
t = s.apply_async()
r = t.get()
assert r == 'foo'