eth-cache

Ethereum chain data caching tools
Info | Log | Files | Refs | LICENSE

test_lmdb.py (3969B)

      1 # standard imports
      2 import unittest
      3 import json
      4 import logging
      5 
      6 # external imports
      7 from chainlib.eth.address import is_same_address
      8 from hexathon import strip_0x
      9 from chainlib.eth.gas import (
     10         Gas,
     11         OverrideGasOracle,
     12         )
     13 from chainlib.eth.nonce import RPCNonceOracle
     14 from chainlib.eth.block import (
     15         block_by_hash,
     16         block_by_number,
     17         Block,
     18         )
     19 from chainlib.eth.tx import (
     20         transaction,
     21         Tx,
     22         )
     23 
     24 
     25 # local imports
     26 from eth_cache.store.lmdb import LmdbStore
     27 from eth_cache.rpc import CacheRPC
     28 from tests.util import TestCache
     29 
     30 logging.basicConfig(level=logging.DEBUG)
     31 logg = logging.getLogger()
     32 
     33 
     34 class TestCacheBasic(TestCache):
     35 
     36     def setUp(self):
     37         super(TestCacheBasic, self).setUp()
     38         self.store = LmdbStore(self.chain_spec, cache_root=self.cache_dir, address_rules=self.address_rules)
     39 
     40 
     41     def test_tx(self):
     42         self.store.put_tx(self.tx, include_data=True)
     43         j = self.store.get_tx(self.tx.hash)
     44         tx = json.loads(j)
     45         self.assertTrue(is_same_address(tx['hash'], self.tx.hash))
     46 
     47 
     48     def test_rcpt(self):
     49         self.store.put_tx(self.tx, include_data=True)
     50         j = self.store.get_rcpt(self.tx.hash)
     51         rcpt = json.loads(j)
     52         self.assertTrue(is_same_address(rcpt['transaction_hash'], self.tx.hash))
     53 
     54 
     55     def test_block_number(self):
     56         self.store.put_block(self.block, include_data=True)
     57         block_hash = strip_0x(self.block.hash)
     58         block_number = int(self.block.number)
     59         j = self.store.get_block_number(block_number)
     60         block = json.loads(j)
     61         retrieved_block_hash = strip_0x(block['hash'])
     62         self.assertEqual(retrieved_block_hash, block_hash)
     63 
     64 
     65     def test_block(self):
     66         self.store.put_block(self.block, include_data=True)
     67         block_hash = strip_0x(self.block.hash)
     68         j = self.store.get_block(block_hash)
     69         block = json.loads(j)
     70         retrieved_block_hash = strip_0x(block['hash'])
     71         self.assertEqual(retrieved_block_hash, block_hash)
     72 
     73 
     74     def test_cache_rpc(self):
     75         rpc = CacheRPC(None, self.store)
     76 
     77         o = block_by_hash(self.block.hash)
     78         block_src = self.rpc.do(o)
     79         self.assertEqual(block_src['hash'], self.block.hash)
     80 
     81         o = block_by_number(self.block.number)
     82         block_src = self.rpc.do(o)
     83         self.assertEqual(block_src['hash'], self.block.hash)
     84 
     85         o = transaction(self.tx.hash)
     86         tx_src = self.rpc.do(o)
     87         self.assertTrue(is_same_address(tx_src['hash'], self.tx.hash))
     88 
     89 
     90     def test_address(self):
     91         nonce_oracle = RPCNonceOracle(self.accounts[2], self.rpc)
     92         gas_oracle = OverrideGasOracle(price=100000000000, limit=30000)
     93         c = Gas(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
     94         (tx_hash, o) = c.create(self.accounts[2], self.accounts[1], 1024)
     95         r = self.rpc.do(o)
     96         o = transaction(tx_hash)
     97         tx_src = self.rpc.do(o)
     98 
     99         o = block_by_hash(tx_src['block_hash'])
    100         block_src = self.rpc.do(o)
    101         block = Block(block_src)
    102 
    103         tx = Tx(tx_src, block=block)
    104         self.store.put_tx(tx, include_data=True)
    105 
    106         nonce_oracle = RPCNonceOracle(self.accounts[1], self.rpc)
    107         c = Gas(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
    108         (tx_hash, o) = c.create(self.accounts[1], self.accounts[0], 1024)
    109         r = self.rpc.do(o)
    110         o = transaction(tx_hash)
    111         tx_src = self.rpc.do(o)
    112 
    113         o = block_by_hash(tx_src['block_hash'])
    114         block_src = self.rpc.do(o)
    115         block = Block(block_src)
    116 
    117         tx = Tx(tx_src, block=block)
    118         self.store.put_tx(tx, include_data=True)
    119 
    120         address = strip_0x(self.accounts[1])
    121         # TODO: add test to limit iteration to within given address
    122         txs = self.store.get_address_tx(address)
    123         self.assertEqual(len(txs), 2)
    124 
    125 
    126 if __name__ == '__main__':
    127     unittest.main()