chainqueue/tests/test_query.py

# standard imports
import os
import logging
import unittest

# 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,
        set_ready,
        set_reserved,
        set_sent,
        set_final,
        )
from chainqueue.enum import StatusBits

# test imports
from tests.chainqueue_base import TestTxBase

logging.basicConfig(level=logging.DEBUG)
logg = logging.getLogger()


class TestTxQuery(TestTxBase):

#    def test_get_tx(self):
#        tx = get_tx(self.chain_spec, self.tx_hash) 
#        expected_keys = [
#                'otx_id',
#                'status',
#                'signed_tx',
#                'nonce',
#                ]
#        for k in tx.keys():
#            expected_keys.remove(k)
#
#        self.assertEqual(len(expected_keys), 0)
#
#
#    def test_nonce_tx(self):
#
#        nonce_hashes = [self.tx_hash]
#        tx_hash = add_0x(os.urandom(32).hex())
#        signed_tx = add_0x(os.urandom(128).hex())
#        create(
#                self.chain_spec,
#                42,
#                self.alice,
#                tx_hash,
#                signed_tx,
#                session=self.session,
#                )
#        txc = TxCache(
#                tx_hash,
#                self.alice,
#                self.bob,
#                self.foo_token,
#                self.bar_token,
#                self.from_value,
#                self.to_value,
#                session=self.session,
#                )
#        self.session.add(txc)
#        self.session.commit()
#
#        nonce_hashes.append(tx_hash)
#
#        tx_hash = add_0x(os.urandom(32).hex())
#        signed_tx = add_0x(os.urandom(128).hex())
#        create(
#                self.chain_spec,
#                41,
#                self.alice,
#                tx_hash,
#                signed_tx,
#                session=self.session,
#                )
#        txc = TxCache(
#                tx_hash,
#                self.alice,
#                self.bob,
#                self.foo_token,
#                self.bar_token,
#                self.from_value,
#                self.to_value,
#                session=self.session,
#                )
#        self.session.add(txc)
#
#        txs = get_nonce_tx_cache(self.chain_spec, 42, self.alice)
#        self.assertEqual(len(txs.keys()), 2)
#    
#        for h in nonce_hashes:
#            self.assertTrue(strip_0x(h) in txs)
#
#
#    def test_paused_tx_cache(self):
#        set_waitforgas(self.chain_spec, self.tx_hash)
#        
#        tx_hash = add_0x(os.urandom(32).hex())
#        signed_tx = add_0x(os.urandom(128).hex())
#        create(
#                self.chain_spec,
#                43,
#                self.alice,
#                tx_hash,
#                signed_tx,
#                session=self.session,
#                )
#        txc = TxCache(
#                tx_hash,
#                self.alice,
#                self.bob,
#                self.foo_token,
#                self.bar_token,
#                self.from_value,
#                self.to_value,
#                session=self.session,
#                )
#        self.session.add(txc)
#        self.session.commit()
#
#        txs = get_paused_tx_cache(self.chain_spec, status=StatusBits.GAS_ISSUES, sender=self.alice, session=self.session)
#        self.assertEqual(len(txs.keys()), 1)
#
#        txs = get_paused_tx_cache(self.chain_spec, status=StatusBits.GAS_ISSUES, session=self.session)
#        self.assertEqual(len(txs.keys()), 1)
#
#        tx_hash = add_0x(os.urandom(32).hex())
#        signed_tx = add_0x(os.urandom(128).hex())
#        create(
#                self.chain_spec,
#                42,
#                self.bob,
#                tx_hash,
#                signed_tx,
#                session=self.session,
#                )
#        txc = TxCache(
#                tx_hash,
#                self.bob,
#                self.alice,
#                self.bar_token,
#                self.foo_token,
#                self.to_value,
#                self.from_value,
#                session=self.session,
#                )
#        self.session.add(txc)
#        self.session.commit()
#
#        txs = get_paused_tx_cache(self.chain_spec, status=StatusBits.GAS_ISSUES, session=self.session)
#        self.assertEqual(len(txs.keys()), 1)
#
#        set_waitforgas(self.chain_spec, tx_hash)
#        self.session.commit()
#
#        txs = get_paused_tx_cache(self.chain_spec, status=StatusBits.GAS_ISSUES, session=self.session)
#        self.assertEqual(len(txs.keys()), 2)
#
#        txs = get_paused_tx_cache(self.chain_spec, status=StatusBits.GAS_ISSUES, sender=self.bob, session=self.session)
#        self.assertEqual(len(txs.keys()), 1)
#
#
#    def test_count(self):
#        for i in range(3):
#            tx_hash = add_0x(os.urandom(32).hex())
#            signed_tx = add_0x(os.urandom(128).hex())
#            create(
#                    self.chain_spec,
#                    i,
#                    self.alice,
#                    tx_hash,
#                    signed_tx,
#                    session=self.session,
#                    )
#            txc = TxCache(
#                tx_hash,
#                self.alice,
#                self.bob,
#                self.foo_token,
#                self.bar_token,
#                self.from_value,
#                self.to_value,
#                session=self.session,
#                )
#            self.session.add(txc)
#            set_ready(self.chain_spec, tx_hash, session=self.session)
#            set_reserved(self.chain_spec, tx_hash, session=self.session)
#            if i > 0:
#                set_sent(self.chain_spec, tx_hash, session=self.session)
#            if i == 2:
#                set_final(self.chain_spec, tx_hash, session=self.session)
#
#        tx_hash = add_0x(os.urandom(32).hex())
#        signed_tx = add_0x(os.urandom(128).hex())
#        create(
#                self.chain_spec,
#                i,
#                self.bob,
#                tx_hash,
#                signed_tx,
#                session=self.session,
#                )
#        txc = TxCache(
#            tx_hash,
#            self.bob,
#            self.carol,
#            self.foo_token,
#            self.bar_token,
#            self.from_value,
#            self.to_value,
#            session=self.session,
#            )
#
#        self.session.add(txc)
#        set_ready(self.chain_spec, tx_hash, session=self.session)
#        set_reserved(self.chain_spec, tx_hash, session=self.session)
#        set_sent(self.chain_spec, tx_hash, session=self.session)
#        self.session.commit()
#
#        self.assertEqual(count_tx(self.chain_spec, status=StatusBits.IN_NETWORK | StatusBits.FINAL, status_target=StatusBits.IN_NETWORK), 2)
#        self.assertEqual(count_tx(self.chain_spec, address=self.alice, status=StatusBits.IN_NETWORK | StatusBits.FINAL, status_target=StatusBits.IN_NETWORK), 1)
#

    def test_account_tx(self):

        nonce_hashes = [self.tx_hash]
        tx_hash = add_0x(os.urandom(32).hex())
        signed_tx = add_0x(os.urandom(128).hex())
        create(
                self.chain_spec,
                42,
                self.alice,
                tx_hash,
                signed_tx,
                session=self.session,
                )
        txc = TxCache(
                tx_hash,
                self.alice,
                self.bob,
                self.foo_token,
                self.bar_token,
                self.from_value,
                self.to_value,
                session=self.session,
                )
        self.session.add(txc)
        self.session.commit()

        nonce_hashes.append(tx_hash)

        time_between = datetime.datetime.utcnow()

        tx_hash = add_0x(os.urandom(32).hex())
        signed_tx = add_0x(os.urandom(128).hex())
        create(
                self.chain_spec,
                41,
                self.alice,
                tx_hash,
                signed_tx,
                session=self.session,
                )
        txc = TxCache(
                tx_hash,
                self.alice,
                self.bob,
                self.foo_token,
                self.bar_token,
                self.from_value,
                self.to_value,
                session=self.session,
                )
        self.session.add(txc)

        nonce_hashes.append(tx_hash)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session)
        self.assertEqual(len(txs.keys()), 3)
         
        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=tx_hash)
        self.assertEqual(len(txs.keys()), 1)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=nonce_hashes[0])
        self.assertEqual(len(txs.keys()), 3)

        bogus_hash = add_0x(os.urandom(32).hex())
        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=bogus_hash)
        self.assertEqual(len(txs.keys()), 0)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=time_between)
        self.assertEqual(len(txs.keys()), 1)

        time_before = time_between - datetime.timedelta(hours=1)
        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=time_before)
        self.assertEqual(len(txs.keys()), 3)

        time_after = datetime.datetime.utcnow()
        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=time_after)
        self.assertEqual(len(txs.keys()), 0)


        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=2)
        self.assertEqual(len(txs.keys()), 2)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=1)
        self.assertEqual(len(txs.keys()), 3)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=4)
        self.assertEqual(len(txs.keys()), 0)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=1, until=2)
        self.assertEqual(len(txs.keys()), 2)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=time_before, until=time_between)
        self.assertEqual(len(txs.keys()), 2)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, session=self.session, since=nonce_hashes[0], until=nonce_hashes[1])
        self.assertEqual(len(txs.keys()), 2)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, status=StatusBits.QUEUED, session=self.session)
        self.assertEqual(len(txs.keys()), 0)

        set_ready(self.chain_spec, nonce_hashes[1], session=self.session)
        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, status=StatusBits.QUEUED, session=self.session)
        self.assertEqual(len(txs.keys()), 1)

        set_reserved(self.chain_spec, nonce_hashes[1], session=self.session)
        set_sent(self.chain_spec, nonce_hashes[1], session=self.session)
        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, status=StatusBits.QUEUED, session=self.session)
        self.assertEqual(len(txs.keys()), 0)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, not_status=StatusBits.QUEUED, session=self.session)
        self.assertEqual(len(txs.keys()), 3)

        txs = get_account_tx(self.chain_spec, self.alice, as_sender=True, as_recipient=False, not_status=StatusBits.QUEUED, status=StatusBits.IN_NETWORK, session=self.session)
        self.assertEqual(len(txs.keys()), 1)


if __name__ == '__main__':
    unittest.main()