chaintool/chainqueue
4
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Rehabilitate tests to new package structure

Browse Source
This commit is contained in:
nolash
2021-05-28 11:13:01 +02:00
parent 18a4d65784
commit 19d4e38dcf
10 changed files with 88 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
tests/base.py
View File

@@ -11,17 +11,20 @@ from chainqueue.db.models.tx import TxCache
from chainlib.chain import ChainSpec
import alembic
import alembic.config
from hexathon import add_0x
from hexathon import (
add_0x,
strip_0x,
)
# local imports
from chainqueue.db import dsn_from_config
from chainqueue.db.models.base import SessionBase
from chainqueue.tx import create
from chainqueue.sql.tx import create
script_dir = os.path.realpath(os.path.dirname(__file__))
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger().getChild(__name__)
#logg = logging.getLogger().getChild(__name__)
logg = logging.getLogger()
class TestBase(unittest.TestCase):
@@ -45,7 +48,7 @@ class TestBase(unittest.TestCase):
SessionBase.poolable = False
SessionBase.transactional = False
SessionBase.procedural = False
SessionBase.connect(dsn, debug=True)
SessionBase.connect(dsn, debug=bool(os.environ.get('DATABASE_DEBUG'))) # TODO: evaluates to "true" even if string is 0
ac = alembic.config.Config(os.path.join(migrationsdir, 'alembic.ini'))
ac.set_main_option('sqlalchemy.url', dsn)
@@ -57,7 +60,7 @@ class TestBase(unittest.TestCase):
self.session = SessionBase.create_session()
self.chain_spec = ChainSpec('evm', 'foo', 42, 'bar')
def tearDown(self):
self.session.commit()
@@ -68,13 +71,16 @@ class TestOtxBase(TestBase):
def setUp(self):
super(TestOtxBase, self).setUp()
self.tx_hash = add_0x(os.urandom(32).hex())
self.tx = add_0x(os.urandom(128).hex())
self.tx_hash = os.urandom(32).hex()
self.tx = os.urandom(128).hex()
self.nonce = 42
self.alice = add_0x(os.urandom(20).hex())
tx_hash = create(self.chain_spec, self.nonce, self.alice, self.tx_hash, self.tx, session=self.session)
self.assertEqual(tx_hash, self.tx_hash)
self.session.commit()
logg.info('using tx hash {}'.format(self.tx_hash))
class TestTxBase(TestOtxBase):

76
tests/test_otx.py
View File

@@ -8,11 +8,11 @@ from chainlib.chain import ChainSpec
# local imports
from chainqueue.db.models.otx import Otx
from chainqueue.state import *
from chainqueue.db.enum import (
is_alive,
is_error_status,
)
from chainqueue.sql.state import *
# test imports
from tests.base import TestOtxBase
@@ -24,7 +24,7 @@ logg = logging.getLogger()
class TestOtx(TestOtxBase):
def test_ideal_state_sequence(self):
set_ready(self.chain_spec, self.tx_hash)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
otx = Otx.load(self.tx_hash, session=self.session)
self.assertEqual(otx.status, StatusBits.QUEUED)
@@ -32,85 +32,85 @@ class TestOtx(TestOtxBase):
self.session.refresh(otx)
self.assertEqual(otx.status, StatusBits.RESERVED)
set_sent(self.chain_spec, self.tx_hash)
set_sent(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status, StatusBits.IN_NETWORK)
set_final(self.chain_spec, self.tx_hash, block=1024)
set_final(self.chain_spec, self.tx_hash, block=1024, session=self.session)
self.session.refresh(otx)
self.assertFalse(is_alive(otx.status))
self.assertFalse(is_error_status(otx.status))
def test_send_fail_and_retry(self):
set_ready(self.chain_spec, self.tx_hash)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
otx = Otx.load(self.tx_hash, session=self.session)
self.assertEqual(otx.status, StatusBits.QUEUED)
set_reserved(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status, StatusBits.RESERVED)
set_sent(self.chain_spec, self.tx_hash, fail=True)
set_sent(self.chain_spec, self.tx_hash, fail=True, session=self.session)
self.session.refresh(otx)
self.assertTrue(is_error_status(otx.status))
set_ready(self.chain_spec, self.tx_hash)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status & StatusBits.QUEUED, StatusBits.QUEUED)
self.assertTrue(is_error_status(otx.status))
set_reserved(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status & StatusBits.RESERVED, StatusBits.RESERVED)
self.assertTrue(is_error_status(otx.status))
set_sent(self.chain_spec, self.tx_hash)
set_sent(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status, StatusBits.IN_NETWORK)
self.assertFalse(is_error_status(otx.status))
set_final(self.chain_spec, self.tx_hash, block=1024)
set_final(self.chain_spec, self.tx_hash, block=1024, session=self.session)
self.session.refresh(otx)
self.assertFalse(is_alive(otx.status))
self.assertFalse(is_error_status(otx.status))
def test_fubar(self):
set_ready(self.chain_spec, self.tx_hash)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
otx = Otx.load(self.tx_hash, session=self.session)
self.assertEqual(otx.status, StatusBits.QUEUED)
set_reserved(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status & StatusBits.RESERVED, StatusBits.RESERVED)
set_fubar(self.chain_spec, self.tx_hash)
set_fubar(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertTrue(is_error_status(otx.status))
self.assertEqual(otx.status & StatusBits.UNKNOWN_ERROR, StatusBits.UNKNOWN_ERROR)
def test_reject(self):
set_ready(self.chain_spec, self.tx_hash)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
otx = Otx.load(self.tx_hash, session=self.session)
self.assertEqual(otx.status, StatusBits.QUEUED)
set_reserved(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status & StatusBits.RESERVED, StatusBits.RESERVED)
set_rejected(self.chain_spec, self.tx_hash)
set_rejected(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertTrue(is_error_status(otx.status))
self.assertEqual(otx.status & StatusBits.NODE_ERROR, StatusBits.NODE_ERROR)
def test_final_fail(self):
set_ready(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash)
set_sent(self.chain_spec, self.tx_hash)
set_final(self.chain_spec, self.tx_hash, block=1042, fail=True)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
set_sent(self.chain_spec, self.tx_hash, session=self.session)
set_final(self.chain_spec, self.tx_hash, block=1042, fail=True, session=self.session)
otx = Otx.load(self.tx_hash, session=self.session)
self.assertFalse(is_alive(otx.status))
self.assertTrue(is_error_status(otx.status))
@@ -118,47 +118,47 @@ class TestOtx(TestOtxBase):
def test_final_protected(self):
set_ready(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash)
set_sent(self.chain_spec, self.tx_hash)
set_final(self.chain_spec, self.tx_hash, block=1042)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
set_sent(self.chain_spec, self.tx_hash, session=self.session)
set_final(self.chain_spec, self.tx_hash, block=1042, session=self.session)
otx = Otx.load(self.tx_hash, session=self.session)
self.assertEqual(otx.status & StatusBits.FINAL, StatusBits.FINAL)
with self.assertRaises(TxStateChangeError):
set_ready(self.chain_spec, self.tx_hash)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
with self.assertRaises(TxStateChangeError):
set_fubar(self.chain_spec, self.tx_hash)
set_fubar(self.chain_spec, self.tx_hash, session=self.session)
with self.assertRaises(TxStateChangeError):
set_rejected(self.chain_spec, self.tx_hash)
set_rejected(self.chain_spec, self.tx_hash, session=self.session)
set_cancel(self.chain_spec, self.tx_hash)
set_cancel(self.chain_spec, self.tx_hash, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status & StatusBits.OBSOLETE, 0)
set_cancel(self.chain_spec, self.tx_hash, manual=True)
set_cancel(self.chain_spec, self.tx_hash, manual=True, session=self.session)
self.session.refresh(otx)
self.assertEqual(otx.status & StatusBits.OBSOLETE, 0)
with self.assertRaises(TxStateChangeError):
set_reserved(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
with self.assertRaises(TxStateChangeError):
set_waitforgas(self.chain_spec, self.tx_hash)
set_waitforgas(self.chain_spec, self.tx_hash, session=self.session)
with self.assertRaises(TxStateChangeError):
set_manual(self.chain_spec, self.tx_hash)
set_manual(self.chain_spec, self.tx_hash, session=self.session)
def test_manual_persist(self):
set_manual(self.chain_spec, self.tx_hash)
set_ready(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash)
set_sent(self.chain_spec, self.tx_hash)
set_final(self.chain_spec, self.tx_hash, block=1042)
set_manual(self.chain_spec, self.tx_hash, session=self.session)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
set_sent(self.chain_spec, self.tx_hash, session=self.session)
set_final(self.chain_spec, self.tx_hash, block=1042, session=self.session)
otx = Otx.load(self.tx_hash, session=self.session)
self.assertEqual(otx.status & StatusBits.MANUAL, StatusBits.MANUAL)

10
tests/test_otx_status_log.py
View File

@@ -3,7 +3,7 @@ import unittest
# local imports
from chainqueue.db.models.otx import Otx
from chainqueue.state import *
from chainqueue.sql.state import *
# test imports
from tests.base import TestOtxBase
@@ -19,10 +19,10 @@ class TestOtxState(TestOtxBase):
def test_state_log(self):
set_ready(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash)
set_sent(self.chain_spec, self.tx_hash)
set_final(self.chain_spec, self.tx_hash, block=1042)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
set_sent(self.chain_spec, self.tx_hash, session=self.session)
set_final(self.chain_spec, self.tx_hash, block=1042, session=self.session)
state_log = get_state_log(self.chain_spec, self.tx_hash)
self.assertEqual(state_log[0][1], StatusEnum.READYSEND)

10
tests/test_query.py
View File

@@ -3,15 +3,17 @@ import os
import logging
import unittest
# local imports
from chainqueue.query import *
from chainqueue.tx import create
from chainqueue.state import set_waitforgas
# external imports
from hexathon import (
add_0x,
strip_0x,
)
# local imports
from chainqueue.sql.query import *
from chainqueue.sql.tx import create
from chainqueue.sql.state import set_waitforgas
# test imports
from tests.base import TestTxBase

17
tests/test_tx_cache.py
View File

@@ -1,11 +1,14 @@
# standard imports
import unittest
# external imports
from hexathon import add_0x
# local imports
from chainqueue.db.models.tx import TxCache
from chainqueue.error import NotLocalTxError
from chainqueue.state import *
from chainqueue.query import get_tx_cache
from chainqueue.sql.state import *
from chainqueue.sql.query import get_tx_cache
# test imports
from tests.base import TestTxBase
@@ -16,10 +19,10 @@ class TestTxCache(TestTxBase):
with self.assertRaises(NotLocalTxError):
TxCache.set_final(self.tx_hash, 1024, 13, session=self.session)
set_ready(self.chain_spec, self.tx_hash)
set_reserved(self.chain_spec, self.tx_hash)
set_sent(self.chain_spec, self.tx_hash)
set_final(self.chain_spec, self.tx_hash, block=1024)
set_ready(self.chain_spec, self.tx_hash, session=self.session)
set_reserved(self.chain_spec, self.tx_hash, session=self.session)
set_sent(self.chain_spec, self.tx_hash, session=self.session)
set_final(self.chain_spec, self.tx_hash, block=1024, session=self.session)
with self.assertRaises(NotLocalTxError):
TxCache.set_final(self.tx_hash, 1023, 13, session=self.session)
@@ -34,7 +37,7 @@ class TestTxCache(TestTxBase):
def test_get(self):
tx_extended_dict = get_tx_cache(self.chain_spec, self.tx_hash)
self.assertEqual(tx_extended_dict['tx_hash'], self.tx_hash)
self.assertEqual(tx_extended_dict['tx_hash'], add_0x(self.tx_hash))
if __name__ == '__main__':