Compare commits

...

8 Commits

3 changed files with 125 additions and 14 deletions

View File

@ -14,9 +14,11 @@ from chainlib.eth.tx import (
)
from chainlib.eth.block import block_by_number
from chainlib.eth.contract import abi_decode_single
from chainlib.eth.constant import ZERO_ADDRESS
from hexathon import strip_0x
from cic_eth_registry import CICRegistry
from cic_eth_registry.erc20 import ERC20Token
from cic_eth_registry.error import UnknownContractError
from chainqueue.db.models.otx import Otx
from chainqueue.db.enum import StatusEnum
from chainqueue.sql.query import get_tx_cache
@ -114,9 +116,6 @@ def list_tx_by_bloom(self, bloomspec, address, chain_spec_dict):
# TODO: pass through registry to validate declarator entry of token
#token = registry.by_address(tx['to'], sender_address=self.call_address)
token = ERC20Token(chain_spec, rpc, tx['to'])
token_symbol = token.symbol
token_decimals = token.decimals
times = tx_times(tx['hash'], chain_spec)
tx_r = {
'hash': tx['hash'],
@ -126,12 +125,6 @@ def list_tx_by_bloom(self, bloomspec, address, chain_spec_dict):
'destination_value': tx_token_value,
'source_token': tx['to'],
'destination_token': tx['to'],
'source_token_symbol': token_symbol,
'destination_token_symbol': token_symbol,
'source_token_decimals': token_decimals,
'destination_token_decimals': token_decimals,
'source_token_chain': chain_str,
'destination_token_chain': chain_str,
'nonce': tx['nonce'],
}
if times['queue'] != None:
@ -146,8 +139,8 @@ def list_tx_by_bloom(self, bloomspec, address, chain_spec_dict):
# TODO: Surely it must be possible to optimize this
# TODO: DRY this with callback filter in cic_eth/runnable/manager
# TODO: Remove redundant fields from end representation (timestamp, tx_hash)
@celery_app.task()
def tx_collate(tx_batches, chain_spec_dict, offset, limit, newest_first=True):
@celery_app.task(bind=True, base=BaseTask)
def tx_collate(self, tx_batches, chain_spec_dict, offset, limit, newest_first=True, verify_contracts=True):
"""Merges transaction data from multiple sources and sorts them in chronological order.
:param tx_batches: Transaction data inputs
@ -196,6 +189,32 @@ def tx_collate(tx_batches, chain_spec_dict, offset, limit, newest_first=True):
if newest_first:
ks.reverse()
for k in ks:
txs.append(txs_by_block[k])
tx = txs_by_block[k]
if verify_contracts:
try:
tx = verify_and_expand(tx, chain_spec, sender_address=BaseTask.call_address)
except UnknownContractError:
logg.error('verify failed on tx {}, skipping'.format(tx['hash']))
continue
txs.append(tx)
return txs
def verify_and_expand(tx, chain_spec, sender_address=ZERO_ADDRESS):
rpc = RPCConnection.connect(chain_spec, 'default')
registry = CICRegistry(chain_spec, rpc)
if tx.get('source_token_symbol') == None and tx['source_token'] != ZERO_ADDRESS:
r = registry.by_address(tx['source_token'], sender_address=sender_address)
token = ERC20Token(chain_spec, rpc, tx['source_token'])
tx['source_token_symbol'] = token.symbol
tx['source_token_decimals'] = token.decimals
if tx.get('destination_token_symbol') == None and tx['destination_token'] != ZERO_ADDRESS:
r = registry.by_address(tx['destination_token'], sender_address=sender_address)
token = ERC20Token(chain_spec, rpc, tx['destination_token'])
tx['destination_token_symbol'] = token.symbol
tx['destination_token_decimals'] = token.decimals
return tx

View File

@ -9,8 +9,8 @@ import semver
version = (
0,
12,
0,
'alpha.3',
1,
'alpha.2',
)
version_object = semver.VersionInfo(

View File

@ -0,0 +1,92 @@
# external imports
import celery
import pytest
from chainlib.connection import RPCConnection
from chainlib.eth.constant import ZERO_ADDRESS
from chainlib.eth.gas import (
RPCGasOracle,
)
from chainlib.eth.tx import (
TxFormat,
unpack,
)
from chainlib.eth.nonce import RPCNonceOracle
from eth_erc20 import ERC20
from hexathon import (
add_0x,
strip_0x,
)
from chainqueue.db.models.tx import TxCache
from chainqueue.db.models.otx import Otx
def test_ext_tx_collate(
default_chain_spec,
init_database,
eth_rpc,
eth_signer,
custodial_roles,
agent_roles,
foo_token,
bar_token,
register_tokens,
cic_registry,
register_lookups,
init_celery_tasks,
celery_session_worker,
):
rpc = RPCConnection.connect(default_chain_spec, 'default')
nonce_oracle = RPCNonceOracle(custodial_roles['FOO_TOKEN_GIFTER'], eth_rpc)
gas_oracle = RPCGasOracle(eth_rpc)
c = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
transfer_value_foo = 1000
transfer_value_bar = 1024
(tx_hash_hex, tx_signed_raw_hex) = c.transfer(foo_token, custodial_roles['FOO_TOKEN_GIFTER'], agent_roles['ALICE'], transfer_value_foo, tx_format=TxFormat.RLP_SIGNED)
tx = unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), default_chain_spec)
otx = Otx(
tx['nonce'],
tx_hash_hex,
tx_signed_raw_hex,
)
init_database.add(otx)
init_database.commit()
txc = TxCache(
tx_hash_hex,
tx['from'],
tx['to'],
foo_token,
bar_token,
transfer_value_foo,
transfer_value_bar,
666,
13,
session=init_database,
)
init_database.add(txc)
init_database.commit()
s = celery.signature(
'cic_eth.ext.tx.tx_collate',
[
{tx_hash_hex: tx_signed_raw_hex},
default_chain_spec.asdict(),
0,
100,
],
queue=None,
)
t = s.apply_async()
r = t.get_leaf()
assert t.successful()
assert len(r) == 1
tx = r[0]
assert tx['source_token_symbol'] == 'FOO'
assert tx['source_token_decimals'] == 6
assert tx['destination_token_symbol'] == 'BAR'
assert tx['destination_token_decimals'] == 9