cic-internal-integration/apps/cic-eth/tests/task/test_task_tx.py
2021-04-02 13:16:27 +00:00

114 lines
3.3 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# standard imports
import logging
# external imports
import pytest
import celery
from chainlib.eth.gas import Gas
from chainlib.eth.nonce import RPCNonceOracle
from chainlib.eth.tx import (
TxFormat,
unpack,
transaction,
receipt,
)
from hexathon import strip_0x
# local imports
from cic_eth.queue.tx import register_tx
from cic_eth.eth.tx import cache_gas_data
from cic_eth.db.models.otx import Otx
logg = logging.getLogger()
@pytest.mark.skip()
def test_tx_send(
init_database,
default_chain_spec,
eth_rpc,
eth_signer,
agent_roles,
contract_roles,
celery_session_worker,
):
chain_id = default_chain_spec.chain_id()
nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], eth_rpc)
c = Gas(signer=eth_signer, nonce_oracle=nonce_oracle, chain_id=chain_id)
(tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 1024, tx_format=TxFormat.RLP_SIGNED)
#unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), chain_id)
register_tx(tx_hash_hex, tx_signed_raw_hex, default_chain_spec, None, session=init_database)
cache_gas_data(tx_hash_hex, tx_signed_raw_hex, default_chain_spec.asdict())
s_send = celery.signature(
'cic_eth.eth.tx.send',
[
[tx_signed_raw_hex],
default_chain_spec.asdict(),
],
queue=None,
)
t = s_send.apply_async()
r = t.get()
assert t.successful()
o = transaction(tx_hash_hex)
tx = eth_rpc.do(o)
assert r == tx['hash']
o = receipt(tx_hash_hex)
rcpt = eth_rpc.do(o)
assert rcpt['status'] == 1
@pytest.mark.skip()
def test_sync_tx(
default_chain_spec,
eth_rpc,
eth_signer,
celery_worker,
):
pass
def test_resend_with_higher_gas(
init_database,
default_chain_spec,
eth_rpc,
eth_signer,
agent_roles,
celery_worker,
):
chain_id = default_chain_spec.chain_id()
nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], eth_rpc)
c = Gas(signer=eth_signer, nonce_oracle=nonce_oracle, chain_id=chain_id)
(tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 1024, tx_format=TxFormat.RLP_SIGNED)
#unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), chain_id)
register_tx(tx_hash_hex, tx_signed_raw_hex, default_chain_spec, None, session=init_database)
cache_gas_data(tx_hash_hex, tx_signed_raw_hex, default_chain_spec.asdict())
tx_before = unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), default_chain_spec.chain_id())
s = celery.signature(
'cic_eth.eth.tx.resend_with_higher_gas',
[
tx_hash_hex,
default_chain_spec.asdict(),
],
queue=None,
)
t = s.apply_async()
r = t.get_leaf()
q = init_database.query(Otx)
q = q.filter(Otx.tx_hash==r)
otx = q.first()
if otx == None:
raise NotLocalTxError(r)
tx_after = unpack(bytes.fromhex(strip_0x(otx.signed_tx)), default_chain_spec.chain_id())
logg.debug('gasprices before {} after {}'.format(tx_before['gasPrice'], tx_after['gasPrice']))
assert tx_after['gasPrice'] > tx_before['gasPrice']