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()