chainqueue

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

test_cache.py (2310B)

      1 # standard imports
      2 import os
      3 import logging
      4 import unittest
      5 import math
      6 
      7 # external imports
      8 from hexathon import add_0x
      9 from chainlib.chain import ChainSpec
     10 
     11 # local imports
     12 from chainqueue import QueueEntry
     13 from chainqueue.cache import (
     14         CacheTokenTx,
     15         CacheFilter,
     16         )
     17 
     18 # test imports
     19 from tests.base_shep import TestShepBase
     20 from tests.common import (
     21         MockTokenCache,
     22         MockCacheTokenTx,
     23         )
     24 
     25 logging.basicConfig(level=logging.DEBUG)
     26 logg = logging.getLogger()
     27 
     28 
     29 class MockNormalizer:
     30 
     31     def address(self, v):
     32         return 'address' + v
     33 
     34 
     35     def value(self, v):
     36         dv = int(math.log10(v) + 1)
     37         return float(v / (10 ** dv))
     38 
     39 
     40     def hash(self, v):
     41         return 'ashbashhash' + v
     42 
     43 
     44 class TestCache(TestShepBase):
     45 
     46     def setUp(self):
     47         super(TestCache, self).setUp()
     48         self.chain_spec = ChainSpec('foo', 'bar', 42, 'baz')
     49         self.cache = MockTokenCache()
     50 
     51     
     52     def test_cache_instance(self):
     53         normalizer = MockNormalizer()
     54         a = b'foo'
     55         tx = MockCacheTokenTx(self.chain_spec, normalizer=normalizer)
     56         tx.deserialize(a)
     57         self.assertTrue(isinstance(tx.value, float))
     58         self.assertEqual(tx.sender[:4], 'addr')
     59         self.assertEqual(tx.recipient[:4], 'addr')
     60         self.assertEqual(tx.hash[:11], 'ashbashhash')
     61 
     62 
     63     def test_cache_putget(self):
     64         a = b'foo'
     65         tx = MockCacheTokenTx(self.chain_spec)
     66         tx.deserialize(a)
     67         self.cache.put(self.chain_spec, tx)
     68         tx_retrieved = self.cache.get(self.chain_spec, tx.hash)
     69         self.assertEqual(tx, tx_retrieved)
     70 
     71 
     72     def test_cache_filter(self):
     73         normalizer = MockNormalizer()
     74         fltr = CacheFilter(normalizer=normalizer)
     75 
     76         sender = os.urandom(20).hex()
     77         fltr.add_senders(sender)
     78 
     79         recipient_one = os.urandom(20).hex()
     80         recipient_two = os.urandom(20).hex()
     81         fltr.add_recipients([recipient_one, recipient_two])
     82 
     83         self.assertEqual(fltr.senders[0][:4], 'addr')
     84         self.assertEqual(fltr.recipients[1][:4], 'addr')
     85 
     86 
     87     def test_cache_query(self):
     88         a = os.urandom(20).hex()
     89         fltr = CacheFilter(nonce=42)
     90         self.cache.count(fltr)
     91         self.assertEqual(self.cache.last_filter, fltr)
     92 
     93 
     94 if __name__ == '__main__':
     95     unittest.main()