chainqueue

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

test_shep.py (1425B)

      1 # standard imports
      2 import os
      3 import logging
      4 import unittest
      5 
      6 # external imports
      7 from hexathon import (
      8         add_0x,
      9         strip_0x,
     10         )
     11 from shep.error import StateTransitionInvalid
     12 
     13 # local imports
     14 from chainqueue import QueueEntry
     15 
     16 # test imports
     17 from tests.base_shep import TestShepBase
     18 from tests.common import MockCacheTokenTx
     19 
     20 logging.basicConfig(level=logging.DEBUG)
     21 logg = logging.getLogger()
     22 
     23 
     24 class TestShep(TestShepBase):
     25    
     26     def test_shep_setup(self):
     27         pass
     28 
     29 
     30     def test_shep_tx(self):
     31         signed_tx = add_0x(os.urandom(128).hex())
     32         nonce = 42
     33         tx = QueueEntry(self.store, cache_adapter=MockCacheTokenTx)
     34         tx_hash = tx.create(signed_tx)
     35 
     36         tx_retrieved = QueueEntry(self.store, tx_hash=tx_hash)
     37         tx_retrieved.load()
     38         self.assertEqual(tx_retrieved.signed_tx, strip_0x(signed_tx))
     39 
     40 
     41     def test_shep_valid(self):
     42         self.state.put('foo', 'bar')
     43         self.state.set('foo', self.state.IN_NETWORK)
     44         self.state.set('foo', self.state.FINAL)
     45     
     46 
     47     def test_shep_invalid(self):
     48         self.state.put('foo', 'bar')
     49         self.state.set('foo', self.state.FINAL)
     50         with self.assertRaises(StateTransitionInvalid):
     51             self.state.move('foo', self.state.INSUFFICIENT_FUNDS)
     52    
     53 
     54     def test_shep_cache(self):
     55         self.store.put('bar', cache_adapter=MockCacheTokenTx)
     56 
     57 
     58 
     59 if __name__ == '__main__':
     60     unittest.main()