Compare commits
4 Commits
Author | SHA1 | Date |
---|---|---|
lash | 5df6656981 | |
lash | 4af735e5d3 | |
lash | 6b6b26f1ae | |
lash | 44bdda80bf |
10
CHANGELOG
10
CHANGELOG
|
@ -1,3 +1,13 @@
|
||||||
|
* 0.4.4
|
||||||
|
- Reinstate tx index bump in sync state on filter execution complete
|
||||||
|
* 0.4.3
|
||||||
|
- Refactor lock code, move into module from runnable script
|
||||||
|
* 0.4.2
|
||||||
|
- Same as 0.4.1
|
||||||
|
* 0.4.1
|
||||||
|
- Correct runnable entry in setup
|
||||||
|
* 0.4.0
|
||||||
|
- Refactor on shep
|
||||||
* 0.3.7
|
* 0.3.7
|
||||||
- Remove hard eth dependency in settings rendering
|
- Remove hard eth dependency in settings rendering
|
||||||
- Add unlock cli tool
|
- Add unlock cli tool
|
||||||
|
|
|
@ -31,16 +31,25 @@ class LockError(Exception):
|
||||||
class FilterDone(Exception):
|
class FilterDone(Exception):
|
||||||
"""Exception raised when all registered filters have been executed
|
"""Exception raised when all registered filters have been executed
|
||||||
"""
|
"""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
class InterruptError(FilterDone):
|
class InterruptError(FilterDone):
|
||||||
"""Exception for interrupting or attempting to use an interrupted sync
|
"""Exception for interrupting or attempting to use an interrupted sync
|
||||||
"""
|
"""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
class IncompleteFilterError(Exception):
|
class IncompleteFilterError(Exception):
|
||||||
"""Exception raised if filter reset is executed prematurely
|
"""Exception raised if filter reset is executed prematurely
|
||||||
"""
|
"""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class FilterInitializationError(BackendError):
|
||||||
|
"""Exception raised if filter state does not match the registered filters
|
||||||
|
"""
|
||||||
|
pass
|
||||||
|
|
||||||
#class AbortTx(Exception):
|
#class AbortTx(Exception):
|
||||||
# """
|
# """
|
||||||
|
|
|
@ -106,40 +106,7 @@ def main():
|
||||||
|
|
||||||
store.connect()
|
store.connect()
|
||||||
store.start(ignore_lock=True)
|
store.start(ignore_lock=True)
|
||||||
|
store.unlock_filter(not action_is_forward)
|
||||||
lock_state = store.filter_state.from_name('LOCK')
|
|
||||||
locked_item = store.filter_state.list(lock_state)
|
|
||||||
if len(locked_item) == 0:
|
|
||||||
sys.stderr.write('Sync filter in {} is not locked\n'.format(state_dir))
|
|
||||||
sys.exit(1)
|
|
||||||
elif len(locked_item) > 1:
|
|
||||||
sys.stderr.write('More than one locked item encountered in {}. That should never happen, so I do not know what to do next.\n'.format(state_dir))
|
|
||||||
sys.exit(1)
|
|
||||||
|
|
||||||
locked_item_key = locked_item[0]
|
|
||||||
locked_item = store.get(int(locked_item_key))
|
|
||||||
locked_state = store.filter_state.state(locked_item_key) - lock_state
|
|
||||||
locked_state_name = store.filter_state.name(locked_state)
|
|
||||||
logg.info('found item "{}" in locked state {}'.format(locked_item, store.filter_state.name(locked_state)))
|
|
||||||
|
|
||||||
if action_is_forward:
|
|
||||||
k = locked_state_name
|
|
||||||
filter_index = None
|
|
||||||
filter_index = filter_list.index(k)
|
|
||||||
filter_pos = filter_index + 1
|
|
||||||
filter_count = len(filter_list)
|
|
||||||
logg.debug('Locked filter {} found at position {} of {}'.format(k, filter_pos, filter_count))
|
|
||||||
if filter_pos == filter_count:
|
|
||||||
logg.info('Locked filter {} is the last filter in the list. Executing filter reset'.format(k))
|
|
||||||
locked_item.reset(check_incomplete=False)
|
|
||||||
else:
|
|
||||||
locked_item.advance(ignore_lock=True)
|
|
||||||
store.filter_state.unset(locked_item_key, lock_state)
|
|
||||||
else:
|
|
||||||
filter_mask = 0xf
|
|
||||||
filter_state = store.filter_state.mask(locked_state, filter_mask)
|
|
||||||
logg.info('Chosen action is "{}": will continue execution at previous filter {}'.format(args.action, store.filter_state.name(filter_state)))
|
|
||||||
store.filter_state.unset(locked_item_key, lock_state)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
|
|
|
@ -13,12 +13,15 @@ class SyncSession:
|
||||||
def __init__(self, session_store):
|
def __init__(self, session_store):
|
||||||
self.session_store = session_store
|
self.session_store = session_store
|
||||||
self.started = self.session_store.started
|
self.started = self.session_store.started
|
||||||
self.get = self.session_store.get
|
|
||||||
self.next = self.session_store.next_item
|
self.next = self.session_store.next_item
|
||||||
self.item = None
|
self.item = None
|
||||||
self.filters = self.session_store.filters
|
self.filters = self.session_store.filters
|
||||||
|
|
||||||
|
|
||||||
|
def get(self, k):
|
||||||
|
return self.session_store.get(str(k))
|
||||||
|
|
||||||
|
|
||||||
def start(self, offset=0, target=-1):
|
def start(self, offset=0, target=-1):
|
||||||
self.session_store.start(offset=offset, target=target)
|
self.session_store.start(offset=offset, target=target)
|
||||||
self.item = self.session_store.next_item()
|
self.item = self.session_store.next_item()
|
||||||
|
|
|
@ -13,6 +13,7 @@ from chainsyncer.error import (
|
||||||
InterruptError,
|
InterruptError,
|
||||||
IncompleteFilterError,
|
IncompleteFilterError,
|
||||||
SyncDone,
|
SyncDone,
|
||||||
|
FilterInitializationError,
|
||||||
)
|
)
|
||||||
|
|
||||||
logg = logging.getLogger(__name__)
|
logg = logging.getLogger(__name__)
|
||||||
|
@ -66,16 +67,6 @@ class SyncItem:
|
||||||
raise FilterDone(self.state_key)
|
raise FilterDone(self.state_key)
|
||||||
|
|
||||||
|
|
||||||
def resume(self):
|
|
||||||
filter_state = self.filter_state.state(self.state_key)
|
|
||||||
if filter_state > 0x0f:
|
|
||||||
filter_state_part = self.filter_state.mask(filter_state, 0x0f)
|
|
||||||
if len(self.filter_state.elements(filter_state)) == 1:
|
|
||||||
logg.info('resume execution on state {} ({})'.format(self.filter_state.name(filter_state_part), filter_state_part))
|
|
||||||
lock_state = self.filter_state.from_name('LOCK')
|
|
||||||
self.filter_state.set(lock_state)
|
|
||||||
|
|
||||||
|
|
||||||
def reset(self, check_incomplete=True):
|
def reset(self, check_incomplete=True):
|
||||||
if check_incomplete:
|
if check_incomplete:
|
||||||
if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
|
if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
|
||||||
|
@ -221,9 +212,9 @@ class SyncStore:
|
||||||
self.state.put(block_number_str, contents=state_bytes)
|
self.state.put(block_number_str, contents=state_bytes)
|
||||||
self.filter_state.put(block_number_str)
|
self.filter_state.put(block_number_str)
|
||||||
o = SyncItem(offset, target, self.state, self.filter_state, ignore_lock=ignore_lock)
|
o = SyncItem(offset, target, self.state, self.filter_state, ignore_lock=ignore_lock)
|
||||||
o.resume()
|
k = str(offset)
|
||||||
self.items[offset] = o
|
self.items[k] = o
|
||||||
self.item_keys.append(offset)
|
self.item_keys.append(k)
|
||||||
elif offset > 0:
|
elif offset > 0:
|
||||||
logg.warning('block number argument {} for start ignored for already initiated sync {}'.format(offset, self.session_id))
|
logg.warning('block number argument {} for start ignored for already initiated sync {}'.format(offset, self.session_id))
|
||||||
self.started = True
|
self.started = True
|
||||||
|
@ -271,9 +262,9 @@ class SyncStore:
|
||||||
if i < lim:
|
if i < lim:
|
||||||
item_target = thresholds[i+1]
|
item_target = thresholds[i+1]
|
||||||
o = SyncItem(block_number, item_target, self.state, self.filter_state, started=True, ignore_lock=ignore_lock)
|
o = SyncItem(block_number, item_target, self.state, self.filter_state, started=True, ignore_lock=ignore_lock)
|
||||||
o.resume()
|
k = str(block_number)
|
||||||
self.items[block_number] = o
|
self.items[k] = o
|
||||||
self.item_keys.append(block_number)
|
self.item_keys.append(k)
|
||||||
logg.info('added existing {}'.format(o))
|
logg.info('added existing {}'.format(o))
|
||||||
|
|
||||||
self.get_target()
|
self.get_target()
|
||||||
|
@ -315,8 +306,65 @@ class SyncStore:
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
|
|
||||||
def peek_next_filter(self):
|
def __get_locked_item(self):
|
||||||
pass
|
locked_item = self.filter_state.list(self.filter_state.state_store.LOCK)
|
||||||
|
|
||||||
|
if len(locked_item) == 0:
|
||||||
|
logg.error('Sync filter in store {} is not locked\n'.format(self))
|
||||||
|
return None
|
||||||
|
elif len(locked_item) > 1:
|
||||||
|
raise FilterInitializationError('More than one locked filter item encountered in store {}. That should never happen, so I do not know what to do next.\n'.format(self))
|
||||||
|
return locked_item[0]
|
||||||
|
|
||||||
|
|
||||||
|
def __get_filter_index(self, k):
|
||||||
|
i = -1
|
||||||
|
fltrs = self.load_filter_list()
|
||||||
|
for fltr in fltrs:
|
||||||
|
i += 1
|
||||||
|
if k == fltr.upper():
|
||||||
|
logg.debug('lock filter match at filter list index {}'.format(i))
|
||||||
|
return (i, fltrs,)
|
||||||
|
|
||||||
|
|
||||||
|
def unlock_filter(self, revert=False):
|
||||||
|
locked_item_key = self.__get_locked_item()
|
||||||
|
if locked_item_key == None:
|
||||||
|
return False
|
||||||
|
locked_item = self.get(locked_item_key)
|
||||||
|
locked_state = self.filter_state.state(locked_item_key) - self.filter_state.state_store.LOCK
|
||||||
|
locked_state_name = self.filter_state.name(locked_state)
|
||||||
|
|
||||||
|
logg.debug('found locked item {} in state {}'.format(locked_item, locked_state))
|
||||||
|
|
||||||
|
(i, fltrs) = self.__get_filter_index(locked_state_name)
|
||||||
|
|
||||||
|
if i == -1:
|
||||||
|
raise FilterInitializationError('locked state {} ({}) found for item {}, but matching filter has not been registered'.format(locked_state_name, locked_state, locked_item))
|
||||||
|
|
||||||
|
if revert:
|
||||||
|
self.__unlock_previous(locked_item, fltrs, i)
|
||||||
|
else:
|
||||||
|
self.__unlock_next(locked_item, fltrs, i)
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def __unlock_next(self, item, lst, index):
|
||||||
|
if index == len(lst) - 1:
|
||||||
|
item.reset(check_incomplete=False)
|
||||||
|
else:
|
||||||
|
item.release()
|
||||||
|
|
||||||
|
|
||||||
|
def __unlock_previous(self, item, lst, index):
|
||||||
|
if index == 0:
|
||||||
|
item.reset(check_incomplete=False)
|
||||||
|
else:
|
||||||
|
new_state_str = lst[index - 1]
|
||||||
|
new_state = self.filter_state.state_store.from_name(new_state_str)
|
||||||
|
self.filter_state.state_store.move(item.state_key, new_state)
|
||||||
|
|
||||||
|
|
||||||
def peek_current_filter(self):
|
def peek_current_filter(self):
|
||||||
pass
|
pass
|
||||||
|
|
|
@ -13,7 +13,8 @@ from chainsyncer.error import NoBlockForYou
|
||||||
from chainsyncer.driver import SyncDriver
|
from chainsyncer.driver import SyncDriver
|
||||||
|
|
||||||
logging.STATETRACE = 5
|
logging.STATETRACE = 5
|
||||||
logg = logging.getLogger().getChild(__name__)
|
logging.addLevelName('STATETRACE', logging.STATETRACE)
|
||||||
|
logg = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def state_event_handler(k, v_old, v_new):
|
def state_event_handler(k, v_old, v_new):
|
||||||
|
|
|
@ -232,7 +232,7 @@ class TestStoreBase(unittest.TestCase):
|
||||||
if self.persist:
|
if self.persist:
|
||||||
store = self.store_factory('foo')
|
store = self.store_factory('foo')
|
||||||
store.start()
|
store.start()
|
||||||
o = store.get(2)
|
o = store.get('2')
|
||||||
|
|
||||||
|
|
||||||
def t_sync_history_interrupted(self):
|
def t_sync_history_interrupted(self):
|
||||||
|
@ -252,7 +252,7 @@ class TestStoreBase(unittest.TestCase):
|
||||||
store.stop(bogus_item)
|
store.stop(bogus_item)
|
||||||
store = self.store_factory('foo')
|
store = self.store_factory('foo')
|
||||||
store.start()
|
store.start()
|
||||||
o = store.get(0)
|
o = store.get('0')
|
||||||
self.assertEqual(o.cursor, 2)
|
self.assertEqual(o.cursor, 2)
|
||||||
self.assertEqual(o.target, 13)
|
self.assertEqual(o.target, 13)
|
||||||
o.next(advance_block=True)
|
o.next(advance_block=True)
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
[metadata]
|
[metadata]
|
||||||
name = chainsyncer
|
name = chainsyncer
|
||||||
version = 0.4.1
|
version = 0.4.4
|
||||||
description = Generic blockchain syncer driver
|
description = Generic blockchain syncer driver
|
||||||
author = Louis Holbrook
|
author = Louis Holbrook
|
||||||
author_email = dev@holbrook.no
|
author_email = dev@holbrook.no
|
||||||
|
@ -22,7 +22,7 @@ licence_files =
|
||||||
|
|
||||||
[options]
|
[options]
|
||||||
include_package_data = True
|
include_package_data = True
|
||||||
python_requires = >= 3.6
|
python_requires = >= 3.7
|
||||||
packages =
|
packages =
|
||||||
chainsyncer
|
chainsyncer
|
||||||
chainsyncer.driver
|
chainsyncer.driver
|
||||||
|
@ -37,5 +37,4 @@ packages =
|
||||||
|
|
||||||
[options.entry_points]
|
[options.entry_points]
|
||||||
console_scripts =
|
console_scripts =
|
||||||
#blocksync-celery = chainsyncer.runnable.tracker:main
|
|
||||||
chainsyncer-unlock = chainsyncer.runnable.unlock:main
|
chainsyncer-unlock = chainsyncer.runnable.unlock:main
|
||||||
|
|
|
@ -19,17 +19,22 @@ from chainsyncer.unittest import (
|
||||||
MockConn,
|
MockConn,
|
||||||
MockTx,
|
MockTx,
|
||||||
MockBlock,
|
MockBlock,
|
||||||
|
MockFilterError,
|
||||||
|
state_event_handler,
|
||||||
|
filter_state_event_handler,
|
||||||
)
|
)
|
||||||
|
|
||||||
logging.basicConfig(level=logging.DEBUG)
|
|
||||||
|
logging.basicConfig(level=logging.STATETRACE)
|
||||||
logg = logging.getLogger()
|
logg = logging.getLogger()
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class TestFilter(unittest.TestCase):
|
class TestFilter(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.path = tempfile.mkdtemp()
|
self.path = tempfile.mkdtemp()
|
||||||
self.store = SyncFsStore(self.path)
|
self.store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
self.session = SyncSession(self.store)
|
self.session = SyncSession(self.store)
|
||||||
self.conn = MockConn()
|
self.conn = MockConn()
|
||||||
|
|
||||||
|
@ -73,6 +78,279 @@ class TestFilter(unittest.TestCase):
|
||||||
self.assertEqual(len(fltr_two.contents), 0)
|
self.assertEqual(len(fltr_two.contents), 0)
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_resume_single_revert(self):
|
||||||
|
fltr_one = MockFilter('foo', brk_hard=True)
|
||||||
|
self.store.register(fltr_one)
|
||||||
|
|
||||||
|
self.session.start()
|
||||||
|
|
||||||
|
item = self.store.get('0')
|
||||||
|
item.next()
|
||||||
|
|
||||||
|
tx_hash = os.urandom(32).hex()
|
||||||
|
tx = MockTx(42, tx_hash)
|
||||||
|
block = MockBlock(13, [tx_hash])
|
||||||
|
|
||||||
|
with self.assertRaises(MockFilterError):
|
||||||
|
self.session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
# Unlock the state, reverting to previous filter
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
self.conn = MockConn()
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
store.connect()
|
||||||
|
store.start(ignore_lock=True)
|
||||||
|
store.unlock_filter(revert=True)
|
||||||
|
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
session = SyncSession(store)
|
||||||
|
self.conn = MockConn()
|
||||||
|
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
|
||||||
|
session.start()
|
||||||
|
|
||||||
|
session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_resume_single_continue(self):
|
||||||
|
fltr_one = MockFilter('foo', brk_hard=True)
|
||||||
|
self.store.register(fltr_one)
|
||||||
|
|
||||||
|
self.session.start()
|
||||||
|
|
||||||
|
item = self.store.get('0')
|
||||||
|
item.next()
|
||||||
|
|
||||||
|
tx_hash = os.urandom(32).hex()
|
||||||
|
tx = MockTx(42, tx_hash)
|
||||||
|
block = MockBlock(13, [tx_hash])
|
||||||
|
|
||||||
|
with self.assertRaises(MockFilterError):
|
||||||
|
self.session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
# Unlock the state, reverting to previous filter
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
self.conn = MockConn()
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
store.connect()
|
||||||
|
store.start(ignore_lock=True)
|
||||||
|
store.unlock_filter(revert=False)
|
||||||
|
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
session = SyncSession(store)
|
||||||
|
self.conn = MockConn()
|
||||||
|
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
store.connect()
|
||||||
|
|
||||||
|
session.start()
|
||||||
|
|
||||||
|
session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_resume_multi_revert_last(self):
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
self.store.register(fltr_one)
|
||||||
|
|
||||||
|
fltr_two = MockFilter('bar', brk_hard=True)
|
||||||
|
self.store.register(fltr_two)
|
||||||
|
|
||||||
|
self.session.start()
|
||||||
|
|
||||||
|
item = self.store.get('0')
|
||||||
|
item.next()
|
||||||
|
|
||||||
|
tx_hash = os.urandom(32).hex()
|
||||||
|
tx = MockTx(42, tx_hash)
|
||||||
|
block = MockBlock(13, [tx_hash])
|
||||||
|
|
||||||
|
with self.assertRaises(MockFilterError):
|
||||||
|
self.session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
# Unlock the state, reverting to previous filter
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
self.conn = MockConn()
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_bar = MockFilter('bar')
|
||||||
|
store.register(fltr_bar)
|
||||||
|
store.connect()
|
||||||
|
store.start(ignore_lock=True)
|
||||||
|
store.unlock_filter(revert=True)
|
||||||
|
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
session = SyncSession(store)
|
||||||
|
self.conn = MockConn()
|
||||||
|
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_two = MockFilter('bar')
|
||||||
|
store.register(fltr_two)
|
||||||
|
|
||||||
|
store.connect()
|
||||||
|
|
||||||
|
session.start()
|
||||||
|
|
||||||
|
session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_resume_multi_continue_last(self):
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
self.store.register(fltr_one)
|
||||||
|
|
||||||
|
fltr_two = MockFilter('bar', brk_hard=True)
|
||||||
|
self.store.register(fltr_two)
|
||||||
|
|
||||||
|
self.session.start()
|
||||||
|
|
||||||
|
item = self.store.get('0')
|
||||||
|
item.next()
|
||||||
|
|
||||||
|
tx_hash = os.urandom(32).hex()
|
||||||
|
tx = MockTx(42, tx_hash)
|
||||||
|
block = MockBlock(13, [tx_hash])
|
||||||
|
|
||||||
|
with self.assertRaises(MockFilterError):
|
||||||
|
self.session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
# Unlock the state, reverting to previous filter
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
self.conn = MockConn()
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_bar = MockFilter('bar')
|
||||||
|
store.register(fltr_bar)
|
||||||
|
store.connect()
|
||||||
|
store.start(ignore_lock=True)
|
||||||
|
store.unlock_filter(revert=False)
|
||||||
|
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
session = SyncSession(store)
|
||||||
|
self.conn = MockConn()
|
||||||
|
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_two = MockFilter('bar')
|
||||||
|
store.register(fltr_two)
|
||||||
|
|
||||||
|
session.start()
|
||||||
|
|
||||||
|
session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_resume_multi_revert_middle(self):
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
self.store.register(fltr_one)
|
||||||
|
|
||||||
|
fltr_two = MockFilter('bar', brk_hard=True)
|
||||||
|
self.store.register(fltr_two)
|
||||||
|
|
||||||
|
fltr_three = MockFilter('baz')
|
||||||
|
self.store.register(fltr_three)
|
||||||
|
|
||||||
|
self.session.start()
|
||||||
|
|
||||||
|
item = self.store.get('0')
|
||||||
|
item.next()
|
||||||
|
|
||||||
|
tx_hash = os.urandom(32).hex()
|
||||||
|
tx = MockTx(42, tx_hash)
|
||||||
|
block = MockBlock(13, [tx_hash])
|
||||||
|
|
||||||
|
with self.assertRaises(MockFilterError):
|
||||||
|
self.session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
# Unlock the state, reverting to previous filter
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
self.conn = MockConn()
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_two = MockFilter('bar')
|
||||||
|
store.register(fltr_two)
|
||||||
|
fltr_three = MockFilter('baz')
|
||||||
|
store.register(fltr_three)
|
||||||
|
|
||||||
|
store.connect()
|
||||||
|
store.start(ignore_lock=True)
|
||||||
|
store.unlock_filter(revert=True)
|
||||||
|
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
session = SyncSession(store)
|
||||||
|
self.conn = MockConn()
|
||||||
|
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_two = MockFilter('bar')
|
||||||
|
store.register(fltr_two)
|
||||||
|
fltr_three = MockFilter('baz')
|
||||||
|
store.register(fltr_three)
|
||||||
|
|
||||||
|
store.connect()
|
||||||
|
|
||||||
|
session.start()
|
||||||
|
|
||||||
|
session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_resume_multi_continue_middle(self):
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
self.store.register(fltr_one)
|
||||||
|
|
||||||
|
fltr_two = MockFilter('bar', brk_hard=True)
|
||||||
|
self.store.register(fltr_two)
|
||||||
|
|
||||||
|
fltr_three = MockFilter('baz')
|
||||||
|
self.store.register(fltr_three)
|
||||||
|
|
||||||
|
self.session.start()
|
||||||
|
|
||||||
|
item = self.store.get('0')
|
||||||
|
item.next()
|
||||||
|
|
||||||
|
tx_hash = os.urandom(32).hex()
|
||||||
|
tx = MockTx(42, tx_hash)
|
||||||
|
block = MockBlock(13, [tx_hash])
|
||||||
|
|
||||||
|
with self.assertRaises(MockFilterError):
|
||||||
|
self.session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
# Unlock the state, reverting to previous filter
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
self.conn = MockConn()
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_two = MockFilter('bar')
|
||||||
|
store.register(fltr_two)
|
||||||
|
fltr_three = MockFilter('baz')
|
||||||
|
store.register(fltr_three)
|
||||||
|
|
||||||
|
store.connect()
|
||||||
|
store.start(ignore_lock=True)
|
||||||
|
store.unlock_filter(revert=False)
|
||||||
|
|
||||||
|
store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
|
||||||
|
session = SyncSession(store)
|
||||||
|
self.conn = MockConn()
|
||||||
|
|
||||||
|
fltr_one = MockFilter('foo')
|
||||||
|
store.register(fltr_one)
|
||||||
|
fltr_two = MockFilter('bar')
|
||||||
|
store.register(fltr_two)
|
||||||
|
fltr_three = MockFilter('baz')
|
||||||
|
store.register(fltr_three)
|
||||||
|
|
||||||
|
session.start()
|
||||||
|
|
||||||
|
session.filter(self.conn, block, tx)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue