taint

Crypto forensics for private use
git clone git://git.defalsify.org/taint.git
Log | Files | Refs | LICENSE

test_syncer.py (2389B)

      1 # standard imports 
      2 import unittest
      3 
      4 # external imports
      5 from chainlib.eth.gas import (
      6         Gas,
      7         RPCGasOracle,
      8     )
      9 from chainlib.eth.nonce import RPCNonceOracle
     10 from chainlib.eth.unittest.ethtester import EthTesterCase
     11 from chainsyncer.driver import HistorySyncer
     12 from chainsyncer.backend.memory import MemBackend
     13 from hexathon import strip_0x
     14 
     15 # local imports
     16 from taint.taint import Tainter
     17 from taint.account import Account
     18 from taint.tag import (
     19         TagPool,
     20         Tag,
     21         )
     22 
     23 
     24 class TestSyncer(EthTesterCase):
     25 
     26     def setUp(self):
     27         super(TestSyncer, self).setUp()
     28         self.syncer_backend = MemBackend(self.chain_spec, None, target_block=10)
     29         self.syncer = HistorySyncer(self.syncer_backend)
     30 
     31         self.alice_account = bytes.fromhex(strip_0x(self.accounts[0]))
     32         self.alice = Account(self.chain_spec, self.alice_account, label='alice')
     33         self.alice.tag(b'foo')
     34 
     35         self.bob_account = bytes.fromhex(strip_0x(self.accounts[1]))
     36         self.bob = Account(self.chain_spec, self.bob_account, label='bob')
     37         self.bob.tag(b'bar')
     38 
     39         self.filter_bits_size = 1024
     40         self.result_handler = TagPool()
     41         self.frontend = Tainter(self.chain_spec, self.filter_bits_size, result_handler=self.result_handler)
     42         self.frontend.add_subject(self.bob)
     43         self.frontend.add_subject(self.alice)
     44 
     45         self.syncer.add_filter(self.frontend)
     46 
     47 
     48     def test_sync_single(self):
     49         self.backend.mine_blocks(3)
     50         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     51         gas_oracle = RPCGasOracle(self.rpc)
     52         c = Gas(self.chain_spec, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle, signer=self.signer)
     53         (tx_hash_hex, o) = c.create(self.accounts[0], self.accounts[1], 1024)
     54         self.rpc.do(o)
     55         self.backend.mine_blocks(3)
     56         
     57         self.assertFalse(self.frontend.have(4, 0))
     58 
     59         self.syncer.loop(0.001, self.rpc)
     60 
     61         self.assertTrue(self.frontend.have(4, 0))
     62 
     63         tag = Tag()
     64         tag.create(b'foo')
     65         tag.create(b'bar')
     66         z = tag.get()
     67 
     68         self.assertEqual(len(self.result_handler.pool.keys()), 3)
     69         self.assertIn(z, self.result_handler.pool.keys())
     70         self.assertIn(self.alice, self.result_handler.pool[z])
     71         self.assertIn(self.bob, self.result_handler.pool[z])
     72 
     73 
     74 if __name__ == '__main__':
     75     unittest.main()