Add fuzzing tests for EventScanner (#138)

* Add fuzzing tests for EventScanner

* Fix poetry check lock

* Set poetry version in CI

* Fix edge case

.github/workflows/ci.yaml

@@ -25,6 +25,7 @@ jobs:
       - name: Install Poetry
         uses: snok/install-poetry@v1
         with:
+          version: 1.8.3
           virtualenvs-create: true
           virtualenvs-in-project: true
           installer-parallel: true
@@ -58,6 +59,7 @@ jobs:
       - name: Install Poetry
         uses: snok/install-poetry@v1
         with:
+          version: 1.8.3
           virtualenvs-create: true
           virtualenvs-in-project: true
           installer-parallel: true
@@ -91,6 +93,7 @@ jobs:
       - name: Install Poetry
         uses: snok/install-poetry@v1
         with:
+          version: 1.8.3
           virtualenvs-create: true
           virtualenvs-in-project: true
           installer-parallel: true

.pre-commit-config.yaml

@@ -66,7 +66,7 @@ repos:
         entry: poetry
         language: system
         files: no-files
-        args: ["lock", "--check"]
+        args: ["check", "--lock"]
         always_run: true
 
   - repo: local

sw_utils/event_scanner.py

@@ -55,9 +55,6 @@ def __init__(
     ):
         self.processor = processor
         self.argument_filters = argument_filters
-        self._contract_call = lambda from_block, to_block: getattr(
-            processor.contract.events, processor.contract_event
-        ).get_logs(argument_filters=argument_filters, fromBlock=from_block, toBlock=to_block)
 
         # Start with half of max chunk size. 1kk chunks works only with powerful nodes.
         start_chunk_size = self.max_scan_chunk_size // 2
@@ -69,7 +66,7 @@ async def process_new_events(self, to_block: BlockNumber) -> None:
         if current_from_block >= to_block:
             return
 
-        while current_from_block < to_block:
+        while current_from_block <= to_block:
             current_to_block, new_events = await self._scan_chunk(current_from_block, to_block)
             await self.processor.process_events(new_events, to_block=current_to_block)
 
@@ -112,6 +109,14 @@ async def _scan_chunk(
 
         raise RuntimeError(f'Failed to sync chunk: from block={from_block}, to block={last_block}')
 
+    async def _contract_call(
+        self, from_block: BlockNumber, to_block: BlockNumber
+    ) -> list[EventData]:
+        event_cls = getattr(self.processor.contract.events, self.processor.contract_event)
+        return await event_cls.get_logs(
+            argument_filters=self.argument_filters, fromBlock=from_block, toBlock=to_block
+        )
+
     def _estimate_next_chunk_size(self) -> None:
         self.chunk_size *= self.chunk_size_multiplier
         self.chunk_size = max(self.min_scan_chunk_size, self.chunk_size)

sw_utils/tests/test_event_scanner.py

@@ -1,11 +1,17 @@
+import random
+from unittest import mock
+
 import pytest
+from attr import dataclass
 from eth_typing import BlockNumber
 from web3.types import EventData
 
 from sw_utils.event_scanner import EventProcessor, EventScanner
 
 
 class MockedEventProcessor(EventProcessor):
+    contract_event = 'event'
+
     @staticmethod
     async def get_from_block() -> BlockNumber:
         return BlockNumber(777)
@@ -23,7 +29,7 @@ async def fetch_events_broken(a, b):
     raise ConnectionError
 
 
-class TestExitSignatureCrud:
+class TestEventScanner:
     async def test_basic(self):
         default_chunk_size = 500000
         p = MockedEventProcessor()
@@ -47,3 +53,99 @@ async def test_basic(self):
         with pytest.raises(ConnectionError):
             await scanner.process_new_events(888)
         assert scanner.chunk_size == default_chunk_size // 8
+
+
+@dataclass
+class EventScannerDB:
+    last_processed_block: BlockNumber | None = None
+    event_blocks: list[int] = []
+
+    def clear(self):
+        self.last_processed_block = None
+        self.event_blocks = []
+
+
+db = EventScannerDB()
+
+
+class SimpleEventProcessor(EventProcessor):
+    contract_event = 'event'
+
+    @staticmethod
+    async def get_from_block() -> BlockNumber:
+        return 0
+
+    @staticmethod
+    async def process_events(events: list[EventData], to_block: BlockNumber) -> None:
+        db.event_blocks.extend([event['blockNumber'] for event in events])
+        db.last_processed_block = to_block
+
+
+class TestEventScannerFuzzing:
+    """
+    Assume that event.get_logs() will raise Exception with 50% probability.
+    Check that all events are scanned and processed.
+    """
+
+    @pytest.mark.parametrize(
+        'min_scan_chunk_size, max_scan_chunk_size, from_block, to_block',
+        [
+            (1, 2, 7, 100),
+            (10, 1000, 700, 10_000),
+        ],
+    )
+    async def test_fuzzing(self, min_scan_chunk_size, max_scan_chunk_size, from_block, to_block):
+        for _ in range(100):
+            try:
+                await self._run_single_test(
+                    min_scan_chunk_size, max_scan_chunk_size, from_block, to_block
+                )
+            finally:
+                db.clear()
+
+    async def _run_single_test(
+        self, min_scan_chunk_size, max_scan_chunk_size, from_block, to_block
+    ):
+        with (
+            mock.patch.object(EventScanner, 'min_scan_chunk_size', min_scan_chunk_size),
+            mock.patch.object(EventScanner, 'max_scan_chunk_size', max_scan_chunk_size),
+            mock.patch.object(SimpleEventProcessor, 'get_from_block', return_value=from_block),
+        ):
+            p = SimpleEventProcessor()
+            scanner = EventScanner(processor=p)
+            event = MockedAsyncEvent()
+            scanner._contract_call = event.fetch_events
+            scanner.request_retry_seconds = 0
+
+            await scanner.process_new_events(to_block=to_block)
+
+            assert db.event_blocks == list(range(from_block, to_block + 1))
+            assert db.last_processed_block == to_block
+
+
+class MockedAsyncEvent:
+    async def fetch_events(self, from_block, to_block) -> list[EventData]:
+        """
+        Raises Exception with 50% probability.
+        Returns list of events otherwise.
+        Single event per block.
+        """
+        is_fail = random.randint(0, 1)
+        if is_fail:
+            raise ConnectionError
+        return [
+            self._get_mocked_event_data(block_number)
+            for block_number in range(from_block, to_block + 1)
+        ]
+
+    def _get_mocked_event_data(self, block_number) -> EventData:
+        return {
+            'address': '0x0',
+            'args': {},
+            'blockHash': '0x0',
+            'blockNumber': block_number,
+            'event': 'event',
+            'logIndex': 0,
+            'transactionHash': '0x0',
+            'transactionIndex': 0,
+        }