cic-stack/apps/cic-eth/tests/unit/db/test_otx.py

98 lines
2.4 KiB
Python

# standard imports
import os
import logging
# third-party imports
import pytest
# local imports
from cic_eth.db.models.base import SessionBase
from cic_eth.db.models.otx import OtxStateLog
from cic_eth.db.models.otx import Otx
from cic_eth.db.enum import (
StatusEnum,
StatusBits,
is_alive,
)
logg = logging.getLogger()
@pytest.mark.skip()
def test_get(
init_w3,
init_database,
):
tx_def = {
'from': init_w3.eth.accounts[0],
'to': init_w3.eth.accounts[1],
'nonce': 0,
'value': 101,
'gasPrice': 2000000000,
'gas': 21000,
'data': '',
'chainId': 1,
}
session = init_database
txs = []
for i in range(10):
nonce = init_w3.eth.getTransactionCount(init_w3.eth.accounts[0], 'pending')
tx_def['nonce'] = nonce
tx = init_w3.eth.sign_transaction(tx_def)
tx_hash = init_w3.eth.send_raw_transaction(tx['raw'])
logg.debug('tx {}'.format(tx))
address = init_w3.eth.accounts[i%3]
otx = Otx(int((i/3)+1), address, '0x'+tx_hash.hex(), tx['raw'])
txs.append(otx)
session.add(otx)
session.flush()
logg.debug(txs)
session.commit()
txs[0].status = 0
session.add(txs[0])
session.commit()
session.close()
get_txs = Otx.get()
logg.debug(get_txs)
def test_state_log(
init_database,
):
Otx.tracing = True
address = '0x' + os.urandom(20).hex()
tx_hash = '0x' + os.urandom(32).hex()
signed_tx = '0x' + os.urandom(128).hex()
otx = Otx.add(0, address, tx_hash, signed_tx, session=init_database)
otx.waitforgas(session=init_database)
init_database.commit()
otx.readysend(session=init_database)
init_database.commit()
otx.sent(session=init_database)
init_database.commit()
otx.success(1024, session=init_database)
init_database.commit()
q = init_database.query(OtxStateLog)
q = q.filter(OtxStateLog.otx_id==otx.id)
q = q.order_by(OtxStateLog.date.asc())
logs = q.all()
assert logs[0].status == StatusEnum.PENDING
assert logs[1].status == StatusEnum.WAITFORGAS
assert logs[2].status & StatusBits.QUEUED
assert logs[3].status & StatusBits.IN_NETWORK
assert not is_alive(logs[4].status)