chainqueue

Blockchain transaction queue control
Info | Log | Files | Refs | LICENSE

chainqueue_base.py (2301B)

      1 # standard imports
      2 import logging
      3 import unittest
      4 import tempfile
      5 import os
      6 #import pysqlite
      7 
      8 # external imports
      9 from chainqueue.db.models.otx import Otx
     10 from chainqueue.db.models.tx import TxCache
     11 from chainlib.chain import ChainSpec
     12 from hexathon import (
     13         add_0x,
     14         strip_0x,
     15         )
     16 
     17 # local imports
     18 from chainqueue.sql.tx import create
     19 from chainqueue.unittest.db import ChainQueueDb
     20 from chainqueue.sql.backend import SQLBackend
     21 
     22 script_dir = os.path.realpath(os.path.dirname(__file__))
     23 
     24 logg = logging.getLogger(__name__)
     25 
     26 
     27 class TestBase(unittest.TestCase):
     28 
     29     def setUp(self):
     30         debug = bool(os.environ.get('DATABASE_DEBUG', False))
     31         self.db = ChainQueueDb(debug=debug)
     32         self.session = self.db.bind_session()
     33         self.chain_spec = ChainSpec('evm', 'foo', 42, 'bar')
     34     
     35 
     36     def tearDown(self):
     37         self.session.commit()
     38         self.db.release_session(self.session)
     39 
     40 
     41 class TestOtxBase(TestBase):
     42 
     43     def setUp(self):
     44         super(TestOtxBase, self).setUp()
     45         self.tx_hash = os.urandom(32).hex()
     46         self.tx = os.urandom(128).hex()
     47         self.nonce = 42
     48         self.alice = add_0x(os.urandom(20).hex())
     49 
     50         tx_hash = create(self.chain_spec, self.nonce, self.alice, self.tx_hash, self.tx, session=self.session)
     51         self.assertEqual(tx_hash, self.tx_hash)
     52         self.session.commit()
     53 
     54         logg.info('using tx hash {}'.format(self.tx_hash))
     55 
     56 
     57 class TestTxBase(TestOtxBase):
     58 
     59     def setUp(self):
     60         super(TestTxBase, self).setUp()
     61         self.bob = add_0x(os.urandom(20).hex())
     62         self.carol = add_0x(os.urandom(20).hex())
     63         self.foo_token = add_0x(os.urandom(20).hex())
     64         self.bar_token = add_0x(os.urandom(20).hex())
     65         self.from_value = 42
     66         self.to_value = 13
     67 
     68         backend = SQLBackend(self.db.dsn)
     69         txc = TxCache(
     70                 self.tx_hash,
     71                 self.alice,
     72                 self.bob,
     73                 self.foo_token,
     74                 self.bar_token,
     75                 self.from_value,
     76                 self.to_value,
     77                 session=self.session,
     78                 )
     79         self.session.add(txc)
     80         self.session.commit()
     81 
     82         otx = Otx.load(self.tx_hash)
     83         txc = TxCache.load(self.tx_hash)
     84 
     85         self.assertEqual(txc.otx_id, otx.id)
     86