chainsyncer

Blockchain syncer driver
Log | Files | Refs | LICENSE

commit 4af735e5d378c152f8d4ca5f58e683396635c1b2
parent 6b6b26f1aefe08f55a3cc3b30bf0444e9dfc9239
Author: lash <dev@holbrook.no>
Date:   Sat, 30 Apr 2022 07:49:10 +0000

Add tests for begin, end, middle revert and continue filter after lock

Diffstat:
Mchainsyncer/runnable/unlock.py | 35-----------------------------------
Mchainsyncer/store/base.py | 21+++------------------
Mtests/test_filter.py | 197+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 200 insertions(+), 53 deletions(-)

diff --git a/chainsyncer/runnable/unlock.py b/chainsyncer/runnable/unlock.py @@ -107,41 +107,6 @@ def main(): store.connect() store.start(ignore_lock=True) store.unlock_filter(not action_is_forward) - return - - 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__': diff --git a/chainsyncer/store/base.py b/chainsyncer/store/base.py @@ -67,19 +67,6 @@ class SyncItem: raise FilterDone(self.state_key) - def resume(self): - return - filter_state = self.filter_state.state(self.state_key) - if filter_state > 0x0f: - filter_state_part = self.filter_state.mask(filter_state, 0x0f) - if filter_state_part > 0: - filter_state_part_name = self.filter_state.name(filter_state_part) - if filter_state_part_name[0] != '_': - 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(self.state_key, lock_state) - - def reset(self, check_incomplete=True): if check_incomplete: if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0: @@ -225,7 +212,6 @@ class SyncStore: self.state.put(block_number_str, contents=state_bytes) self.filter_state.put(block_number_str) o = SyncItem(offset, target, self.state, self.filter_state, ignore_lock=ignore_lock) - o.resume() k = str(offset) self.items[k] = o self.item_keys.append(k) @@ -276,7 +262,6 @@ class SyncStore: if i < lim: item_target = thresholds[i+1] 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[k] = o self.item_keys.append(k) @@ -366,7 +351,6 @@ class SyncStore: def __unlock_next(self, item, lst, index): - #self.filter_state.state_store.unset(item.state_key, self.filter_state.state_store.LOCK) if index == len(lst) - 1: item.reset(check_incomplete=False) else: @@ -377,8 +361,9 @@ class SyncStore: if index == 0: item.reset(check_incomplete=False) else: - new_state = lst(index) - self.filter_state.state_store.from_name(new_state.upper()) + 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): diff --git a/tests/test_filter.py b/tests/test_filter.py @@ -154,6 +154,203 @@ class TestFilter(unittest.TestCase): 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__': unittest.main()