eth-cache

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

util.py (2080B)

      1 # standard imports
      2 import tempfile
      3 import shutil
      4 import unittest
      5 
      6 # external imports
      7 from chainlib.chain import ChainSpec
      8 from chainlib.eth.gas import (
      9         Gas,
     10         OverrideGasOracle,
     11         )
     12 from chainlib.eth.nonce import RPCNonceOracle
     13 from chainlib.eth.tx import (
     14         transaction,
     15         receipt,
     16         TxFormat,
     17         Tx,
     18         )
     19 from chainlib.eth.block import (
     20         block_by_hash,
     21         Block,
     22         )
     23 from chainlib.eth.dialect import DialectFilter as EthDialectFilter
     24 
     25 from chainlib.eth.unittest.ethtester import EthTesterCase
     26 
     27 # TODO: 2024-06-16 suddenly state_root value is binary, which breaks the json serializer. this seems to be a bug in the eth-tester, and this code should probably be moved to chainlib-eth unittest and implemented by default in test cases
     28 class DialectFilter(EthDialectFilter):
     29     
     30     def apply_src(self, src, dialect_filter=None):
     31 
     32         for k in src.keys():
     33             if type(src[k]) == bytes:
     34                 src[k] = '0x' + src[k].hex()
     35 
     36         return src
     37 
     38 
     39 class TestCache(EthTesterCase):
     40 
     41     def setUp(self):
     42         super(TestCache, self).setUp()
     43         fp = tempfile.mkdtemp()
     44         self.cache_dir = fp
     45 
     46         class Applier:
     47             def apply_rules_addresses(self, sender, recipient, address):
     48                 return True
     49 
     50         self.address_rules = Applier()
     51         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     52         gas_oracle = OverrideGasOracle(price=100000000000, limit=30000)
     53         c = Gas(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
     54         (tx_hash, o) = c.create(self.accounts[0], self.accounts[1], 1024)
     55         r = self.rpc.do(o)
     56 
     57         o = transaction(tx_hash)
     58         tx_src = self.rpc.do(o)
     59 
     60         o = receipt(tx_hash)
     61         rcpt_src = self.rpc.do(o)
     62 
     63         o = block_by_hash(tx_src['block_hash'])
     64         block_src = self.rpc.do(o)
     65 
     66         fltr = DialectFilter()
     67         self.block = Block(block_src, dialect_filter=fltr)
     68         self.tx = Tx(tx_src, block=self.block, rcpt=rcpt_src, dialect_filter=fltr)