chainsyncer

Blockchain syncer driver
Info | Log | Files | Refs | LICENSE

test_driver.py (1426B)

      1 # standard imports
      2 import unittest
      3 import tempfile
      4 import shutil
      5 import logging
      6 import stat
      7 import os
      8 
      9 # local imports
     10 from chainsyncer.store.fs import SyncFsStore
     11 from chainsyncer.session import SyncSession
     12 from chainsyncer.error import (
     13         LockError,
     14         FilterDone,
     15         IncompleteFilterError,
     16         SyncDone,
     17         )
     18 from chainsyncer.unittest import (
     19         MockBlockGenerator,
     20         MockFilter,
     21         MockChainInterfaceConn,
     22         MockTx,
     23         MockBlock,
     24         MockChainInterface,
     25         MockFilterError,
     26         )
     27 from chainsyncer.driver.chain_interface import ChainInterfaceDriver
     28 
     29 logging.basicConfig(level=logging.DEBUG)
     30 logg = logging.getLogger()
     31 
     32 
     33 class TestFilter(unittest.TestCase):
     34 
     35     def setUp(self):
     36         self.path = tempfile.mkdtemp()
     37         self.store = SyncFsStore(self.path)
     38         self.ifc = MockChainInterface()
     39         self.conn = MockChainInterfaceConn(self.ifc)
     40 
     41 
     42     def tearDown(self):
     43         shutil.rmtree(self.path)
     44 
     45 
     46     def test_driver(self):
     47         generator = MockBlockGenerator()
     48         generator.generate([1, 2], driver=self.conn)
     49 
     50         drv = ChainInterfaceDriver(self.store, self.ifc, target=1)
     51 
     52         fltr_one = MockFilter('foo')
     53         self.store.register(fltr_one)
     54         with self.assertRaises(SyncDone):
     55             drv.run(self.conn)
     56 
     57         self.assertEqual(len(fltr_one.contents), 3)
     58 
     59 
     60 if __name__ == '__main__':
     61     unittest.main()