eth-cache

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

test_basic.py (4004B)

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