cic-internal-integration/apps/cic-eth/tests/task/test_task_admin.py
2021-06-03 13:51:55 +00:00

89 lines
2.6 KiB
Python
Raw Permalink 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 celery
from chainlib.connection import RPCConnection
from chainlib.eth.nonce import OverrideNonceOracle
from chainqueue.sql.tx import (
create as queue_create,
)
from chainlib.eth.gas import (
Gas,
OverrideGasOracle,
)
from chainlib.eth.tx import TxFormat
from chainqueue.sql.query import get_nonce_tx_cache
from chainqueue.db.models.otx import Otx
from chainqueue.db.enum import StatusBits
from hexathon import add_0x
# local imports
from cic_eth.admin.nonce import shift_nonce
from cic_eth.eth.gas import cache_gas_data
logg = logging.getLogger()
def test_shift_nonce(
default_chain_spec,
init_database,
eth_rpc,
eth_signer,
agent_roles,
celery_session_worker,
caplog,
):
nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
gas_oracle = OverrideGasOracle(limit=21000, conn=eth_rpc)
tx_hashes = []
txs = []
for i in range(10):
c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
(tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
queue_create(
default_chain_spec,
42+i,
agent_roles['ALICE'],
tx_hash_hex,
tx_signed_raw_hex,
session=init_database,
)
cache_gas_data(
tx_hash_hex,
tx_signed_raw_hex,
default_chain_spec.asdict(),
)
tx_hashes.append(tx_hash_hex)
txs.append(tx_signed_raw_hex)
init_database.commit()
s = celery.signature(
'cic_eth.admin.nonce.shift_nonce',
[
default_chain_spec.asdict(),
tx_hashes[3],
],
queue=None
)
t = s.apply_async()
r = t.get_leaf()
assert t.successful()
init_database.commit()
for i in range(42+3, 42+10):
txs = get_nonce_tx_cache(default_chain_spec, i, agent_roles['ALICE'], session=init_database)
for k in txs.keys():
hsh = add_0x(k)
otx = Otx.load(hsh, session=init_database)
logg.debug('checking nonce {} tx {} status {}'.format(i, otx.tx_hash, otx.status))
if add_0x(k) == tx_hashes[i-42]:
assert otx.status & StatusBits.OBSOLETE == StatusBits.OBSOLETE
else:
assert otx.status == 1