cic-stack/apps/cic-eth/tests/task/test_task_account.py

189 lines
4.6 KiB
Python
Raw Normal View History

2021-02-01 18:12:51 +01:00
# standard imports
import os
import logging
import time
# third-party imports
import pytest
import celery
from chainlib.connection import RPCConnection
from chainlib.eth.nonce import RPCNonceOracle
from chainlib.eth.tx import receipt
from eth_accounts_index import AccountRegistry
from hexathon import strip_0x
2021-02-01 18:12:51 +01:00
# 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
logg = logging.getLogger()
2021-02-01 18:12:51 +01:00
def test_create_account(
default_chain_spec,
eth_rpc,
2021-02-01 18:12:51 +01:00
init_database,
celery_session_worker,
caplog,
2021-02-01 18:12:51 +01:00
):
s = celery.signature(
'cic_eth.eth.account.create',
[
'foo',
default_chain_spec.asdict(),
2021-02-01 18:12:51 +01:00
],
)
t = s.apply_async()
r = t.get()
session = SessionBase.create_session()
q = session.query(Nonce).filter(Nonce.address_hex==r)
o = q.first()
session.close()
assert o != None
assert o.nonce == 0
s = celery.signature(
'cic_eth.eth.account.have',
[
r,
default_chain_spec.asdict(),
2021-02-01 18:12:51 +01:00
],
)
t = s.apply_async()
assert r == t.get()
def test_register_account(
default_chain_spec,
account_registry,
2021-02-01 18:12:51 +01:00
init_database,
init_eth_tester,
eth_accounts,
eth_rpc,
2021-02-01 18:12:51 +01:00
cic_registry,
eth_empty_accounts,
custodial_roles,
call_sender,
celery_session_worker,
2021-02-01 18:12:51 +01:00
):
2021-03-06 18:55:51 +01:00
s_nonce = celery.signature(
'cic_eth.eth.tx.reserve_nonce',
2021-02-01 18:12:51 +01:00
[
eth_empty_accounts[0],
custodial_roles['ACCOUNT_REGISTRY_WRITER'],
2021-03-06 18:55:51 +01:00
],
queue=None,
)
s_register = celery.signature(
'cic_eth.eth.account.register',
[
default_chain_spec.asdict(),
custodial_roles['ACCOUNT_REGISTRY_WRITER'],
2021-02-01 18:12:51 +01:00
],
queue=None,
2021-02-01 18:12:51 +01:00
)
2021-03-06 18:55:51 +01:00
s_nonce.link(s_register)
t = s_nonce.apply_async()
2021-02-01 18:12:51 +01:00
address = t.get()
2021-03-06 18:55:51 +01:00
for r in t.collect():
logg.debug('r {}'.format(r))
2021-03-06 18:55:51 +01:00
assert t.successful()
2021-02-01 18:12:51 +01:00
session = SessionBase.create_session()
o = session.query(Otx).first()
tx_signed_hex = o.signed_tx
session.close()
2021-02-01 18:12:51 +01:00
s_send = celery.signature(
'cic_eth.eth.tx.send',
[
[tx_signed_hex],
default_chain_spec.asdict(),
2021-02-01 18:12:51 +01:00
],
queue=None,
2021-02-01 18:12:51 +01:00
)
t = s_send.apply_async()
address = t.get()
r = t.collect()
t.successful()
init_eth_tester.mine_block()
c = AccountRegistry()
o = c.have(account_registry, eth_empty_accounts[0], sender_address=call_sender)
r = eth_rpc.do(o)
assert int(strip_0x(r), 16) == 1
2021-02-01 18:12:51 +01:00
def test_role_task(
default_chain_spec,
2021-02-01 18:12:51 +01:00
init_database,
celery_session_worker,
):
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,
default_chain_spec.asdict(),
2021-02-01 18:12:51 +01:00
],
)
t = s.apply_async()
r = t.get()
assert r == 'foo'
def test_gift(
init_database,
default_chain_spec,
contract_roles,
agent_roles,
account_registry,
faucet,
eth_rpc,
eth_signer,
init_celery_tasks,
cic_registry,
celery_session_worker,
):
nonce_oracle = RPCNonceOracle(contract_roles['ACCOUNT_REGISTRY_WRITER'], eth_rpc)
c = AccountRegistry(signer=eth_signer, nonce_oracle=nonce_oracle, chain_id=default_chain_spec.chain_id())
(tx_hash_hex, o) = c.add(account_registry, contract_roles['ACCOUNT_REGISTRY_WRITER'], agent_roles['ALICE'])
eth_rpc.do(o)
o = receipt(tx_hash_hex)
r = eth_rpc.do(o)
assert r['status'] == 1
s_nonce = celery.signature(
'cic_eth.eth.tx.reserve_nonce',
[
agent_roles['ALICE'],
],
queue=None,
)
s_gift = celery.signature(
'cic_eth.eth.account.gift',
[
default_chain_spec.asdict(),
],
queue=None,
)
s_nonce.link(s_gift)
t = s_nonce.apply_async()
r = t.get_leaf()
assert t.successful()