175 lines
4.3 KiB
Python
175 lines
4.3 KiB
Python
# standard imports
|
|
import os
|
|
import logging
|
|
|
|
# third-party imports
|
|
import pytest
|
|
import celery
|
|
from cic_registry import zero_address
|
|
|
|
# local imports
|
|
from cic_eth.db.models.otx import Otx
|
|
from cic_eth.db.models.tx import TxCache
|
|
from cic_eth.db.enum import StatusEnum
|
|
|
|
logg = logging.getLogger()
|
|
|
|
# TODO: Refactor to use test vector decorator
|
|
def test_status_success(
|
|
init_w3,
|
|
init_database,
|
|
celery_session_worker,
|
|
):
|
|
|
|
tx_hash = '0x' + os.urandom(32).hex()
|
|
signed_tx = '0x' + os.urandom(128).hex()
|
|
account = '0x' + os.urandom(20).hex()
|
|
|
|
otx = Otx(0, init_w3.eth.accounts[0], tx_hash, signed_tx)
|
|
init_database.add(otx)
|
|
init_database.commit()
|
|
assert otx.status == StatusEnum.PENDING
|
|
|
|
txc = TxCache(tx_hash, account, init_w3.eth.accounts[0], zero_address, zero_address, 13, 13)
|
|
init_database.add(txc)
|
|
init_database.commit()
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_sent_status',
|
|
[tx_hash],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.SENT
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_final_status',
|
|
[tx_hash, 13],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.SUCCESS
|
|
|
|
|
|
def test_status_tempfail_resend(
|
|
init_w3,
|
|
init_database,
|
|
celery_session_worker,
|
|
):
|
|
|
|
tx_hash = '0x' + os.urandom(32).hex()
|
|
signed_tx = '0x' + os.urandom(128).hex()
|
|
account = '0x' + os.urandom(20).hex()
|
|
|
|
otx = Otx(0, init_w3.eth.accounts[0], tx_hash, signed_tx)
|
|
init_database.add(otx)
|
|
init_database.commit()
|
|
|
|
txc = TxCache(tx_hash, account, init_w3.eth.accounts[0], zero_address, zero_address, 13, 13)
|
|
init_database.add(txc)
|
|
init_database.commit()
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_sent_status',
|
|
[tx_hash, True],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.SENDFAIL
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_sent_status',
|
|
[tx_hash],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.SENT
|
|
|
|
|
|
|
|
def test_status_fail(
|
|
init_w3,
|
|
init_database,
|
|
celery_session_worker,
|
|
):
|
|
|
|
tx_hash = '0x' + os.urandom(32).hex()
|
|
signed_tx = '0x' + os.urandom(128).hex()
|
|
account = '0x' + os.urandom(20).hex()
|
|
|
|
otx = Otx(0, init_w3.eth.accounts[0], tx_hash, signed_tx)
|
|
init_database.add(otx)
|
|
init_database.commit()
|
|
|
|
txc = TxCache(tx_hash, account, init_w3.eth.accounts[0], zero_address, zero_address, 13, 13)
|
|
init_database.add(txc)
|
|
init_database.commit()
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_sent_status',
|
|
[tx_hash],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.SENT
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_final_status',
|
|
[tx_hash, 13, True],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.REVERTED
|
|
|
|
|
|
|
|
def test_status_fubar(
|
|
init_w3,
|
|
init_database,
|
|
celery_session_worker,
|
|
):
|
|
|
|
tx_hash = '0x' + os.urandom(32).hex()
|
|
signed_tx = '0x' + os.urandom(128).hex()
|
|
account = '0x' + os.urandom(20).hex()
|
|
|
|
otx = Otx(0, init_w3.eth.accounts[0], tx_hash, signed_tx)
|
|
init_database.add(otx)
|
|
init_database.commit()
|
|
|
|
txc = TxCache(tx_hash, account, init_w3.eth.accounts[0], zero_address, zero_address, 13, 13)
|
|
init_database.add(txc)
|
|
init_database.commit()
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_sent_status',
|
|
[tx_hash],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.SENT
|
|
|
|
s = celery.signature(
|
|
'cic_eth.queue.tx.set_fubar',
|
|
[tx_hash],
|
|
)
|
|
t = s.apply_async()
|
|
t.get()
|
|
assert t.successful()
|
|
init_database.refresh(otx)
|
|
assert otx.status == StatusEnum.FUBAR
|