test: Account for Unpredictability in Block Range Received by Underlying Subscriber (#188)

Author: 0xNeshi

src/test_utils/macros.rs

@@ -1,3 +1,8 @@
+use alloy::primitives::LogData;
+use tokio_stream::Stream;
+
+use crate::Message;
+
 #[macro_export]
 macro_rules! assert_next {
     ($stream: expr, $expected: expr) => {
@@ -13,7 +18,7 @@ macro_rules! assert_next {
         if let Some(msg) = message {
             assert_eq!(msg, $expected)
         } else {
-            panic!("Expected {:?}, got: {message:?}", $expected)
+            panic!("Expected {:?}, but channel was closed", $expected)
         }
     };
 }
@@ -43,6 +48,173 @@ macro_rules! assert_empty {
     }};
 }
 
+/// Asserts that a stream emits a specific sequence of events in order.
+///
+/// This macro consumes messages from a stream and verifies that the provided events are emitted
+/// in the exact order specified, regardless of how they are batched. The stream may emit events
+/// across multiple batches or all at once—the macro handles both cases. It ensures no unexpected
+/// events appear between the expected ones and that the sequence completes exactly as specified.
+///
+/// The macro accepts events of any type implementing [`SolEvent`](alloy::sol_types::SolEvent).
+/// Events are compared by their encoded log data, allowing flexible matching across different
+/// batch configurations while maintaining strict ordering requirements.
+///
+/// # Examples
+///
+/// ```no_run
+/// # use alloy::sol;
+/// sol! {
+///     event CountIncreased(uint256 newCount);
+/// }
+///
+/// #[tokio::test]
+/// async fn test_event_order() {
+///     // scanner setup...
+///
+///     let mut stream = scanner.subscribe(EventFilter::new().contract_address(contract_address));
+///
+///     // Assert these two events are emitted in order
+///     assert_event_sequence!(
+///         stream,
+///         &[
+///             CountIncreased { newCount: U256::from(1) },
+///             CountIncreased { newCount: U256::from(2) },
+///         ]
+///     );
+/// }
+/// ```
+///
+/// The assertion passes whether events arrive in separate batches or together:
+/// * **Separate batches**: `[Event1]`, then `[Event2]`
+/// * **Single batch**: `[Event1, Event2]`
+///
+/// # Panics
+///
+/// * **Timeout**: The stream doesn't produce the next expected event within the timeout period
+///   (default 5 seconds, configurable via `timeout = N` parameter).
+/// * **Wrong event**: The stream emits a different event than the next expected one in the
+///   sequence.
+/// * **Extra events**: The stream emits more events than expected after the sequence completes.
+/// * **Stream closed early**: The stream ends before all expected events are emitted.
+/// * **Wrong message type**: The stream yields a non-`Data` message (e.g., `Error` or `Status`)
+///   when an event is expected.
+/// * **Empty sequence**: The macro is called with an empty event collection (use `assert_empty!`
+///   instead).
+///
+/// On panic, the error message includes the remaining expected events for debugging.
+#[macro_export]
+macro_rules! assert_event_sequence {
+    // owned slices just pass to the borrowed slices variant
+    ($stream: expr, [$($event:expr),+ $(,)?]) => {
+        assert_event_sequence!($stream, &[$($event),+], timeout = 5)
+    };
+    ($stream: expr, [$($event:expr),+ $(,)?], timeout = $secs: expr) => {
+        assert_event_sequence!($stream, &[$($event),+], timeout = $secs)
+    };
+    // borrowed slices
+    ($stream: expr, &[$($event:expr),+ $(,)?]) => {
+        assert_event_sequence!($stream, &[$($event),+], timeout = 5)
+    };
+    ($stream: expr, &[$($event:expr),+ $(,)?], timeout = $secs: expr) => {
+        let expected_options = &[$(alloy::sol_types::SolEvent::encode_log_data(&$event)),+];
+
+       $crate::test_utils::macros::assert_event_sequence(&mut $stream, expected_options, $secs).await
+    };
+    // variables and non-slice expressions
+    ($stream: expr, $events: expr) => {
+        assert_event_sequence!($stream, $events, timeout = 5)
+    };
+    ($stream: expr, $events: expr, timeout = $secs: expr) => {
+        let expected_options = $events.iter().map(alloy::sol_types::SolEvent::encode_log_data).collect::<Vec<_>>();
+        if expected_options.is_empty() {
+            panic!("error: assert_event_sequence! called with an empty collection. Use assert_empty! macro instead to check for no pending messages.")
+        }
+        $crate::test_utils::macros::assert_event_sequence(&mut $stream, expected_options.iter(), $secs).await
+    };
+}
+
+/// Same as [`assert_event_sequence!`], but invokes [`assert_empty!`] at the end.
+#[macro_export]
+macro_rules! assert_event_sequence_final {
+    // owned slices
+    ($stream: expr, [$($event:expr),+ $(,)?]) => {{
+        assert_event_sequence_final!($stream, &[$($event),+])
+    }};
+    ($stream: expr, [$($event:expr),+ $(,)?], timeout = $secs: expr) => {{
+        assert_event_sequence_final!($stream, &[$($event),+], timeout = $secs)
+    }};
+    // borrowed slices
+    ($stream: expr, &[$($event:expr),+ $(,)?]) => {{
+        assert_event_sequence_final!($stream, &[$($event),+], timeout = 5)
+    }};
+    ($stream: expr, &[$($event:expr),+ $(,)?], timeout = $secs: expr) => {{
+        $crate::assert_event_sequence!($stream, &[$($event),+], timeout = $secs);
+        $crate::assert_empty!($stream)
+    }};
+    // variables and non-slice expressions
+    ($stream: expr, $events: expr) => {{
+        assert_event_sequence_final!($stream, $events, timeout = 5)
+    }};
+    ($stream: expr, $events: expr, timeout = $secs: expr) => {{
+        $crate::assert_event_sequence!($stream, $events, timeout = $secs);
+        $crate::assert_empty!($stream)
+    }};
+}
+
+#[allow(clippy::missing_panics_doc)]
+pub async fn assert_event_sequence<S: Stream<Item = Message> + Unpin>(
+    stream: &mut S,
+    expected_options: impl IntoIterator<Item = &LogData>,
+    timeout_secs: u64,
+) {
+    let mut remaining = expected_options.into_iter();
+    let start = std::time::Instant::now();
+    let timeout_duration = std::time::Duration::from_secs(timeout_secs);
+
+    while let Some(expected) = remaining.next() {
+        let elapsed = start.elapsed();
+
+        assert!(
+            elapsed < timeout_duration,
+            "Timed out waiting for events. Still expecting: {:#?}",
+            remaining.collect::<Vec<_>>()
+        );
+
+        let time_left = timeout_duration - elapsed;
+        let message = tokio::time::timeout(time_left, tokio_stream::StreamExt::next(stream))
+            .await
+            .expect("timed out waiting for next batch");
+
+        match message {
+            Some(Message::Data(batch)) => {
+                let mut batch = batch.iter();
+                let event = batch.next().expect("Streamed batch should not be empty");
+                assert_eq!(
+                    expected,
+                    event.data(),
+                    "\nRemaining: {:#?}\n",
+                    remaining.collect::<Vec<_>>()
+                );
+                while let Some(event) = batch.next() {
+                    let expected = remaining.next().unwrap_or_else(|| panic!("Received more events than expected.\nNext event: {:#?}\nStreamed remaining: {batch:#?}", event.data()));
+                    assert_eq!(
+                        expected,
+                        event.data(),
+                        "\nRemaining: {:#?}\n",
+                        remaining.collect::<Vec<_>>()
+                    );
+                }
+            }
+            Some(other) => {
+                panic!("Expected Message::Data, got: {other:#?}");
+            }
+            None => {
+                panic!("Stream closed while still expecting: {:#?}", remaining.collect::<Vec<_>>());
+            }
+        }
+    }
+}
+
 /// Asserts that a stream of block ranges completely covers an expected block range.
 ///
 /// This macro consumes messages from a stream and verifies that the block ranges received
@@ -154,3 +326,35 @@ macro_rules! assert_range_coverage {
         }
     }};
 }
+
+#[cfg(test)]
+mod tests {
+    use alloy::sol;
+    use tokio::sync::mpsc;
+    use tokio_stream::wrappers::ReceiverStream;
+
+    sol! {
+        #[derive(Debug)]
+        event Transfer(address indexed from, address indexed to, uint256 value);
+    }
+
+    #[tokio::test]
+    #[should_panic = "error: assert_event_sequence! called with an empty collection. Use assert_empty! macro instead to check for no pending messages."]
+    async fn assert_event_sequence_macro_with_empty_vec() {
+        let (_tx, rx) = mpsc::channel(10);
+        let mut stream = ReceiverStream::new(rx);
+
+        let empty_vec: Vec<Transfer> = Vec::new();
+        assert_event_sequence!(stream, empty_vec);
+    }
+
+    #[tokio::test]
+    #[should_panic = "error: assert_event_sequence! called with an empty collection. Use assert_empty! macro instead to check for no pending messages."]
+    async fn assert_event_sequence_macro_with_empty_slice() {
+        let (_tx, rx) = mpsc::channel(10);
+        let mut stream = ReceiverStream::new(rx);
+
+        let empty_vec: &[Transfer] = &[];
+        assert_event_sequence!(stream, empty_vec);
+    }
+}

tests/historic/basic.rs

@@ -5,13 +5,9 @@ use crate::common::{TestCounter, setup_historic_scanner};
 
 #[tokio::test]
 async fn processes_events_within_specified_historical_range() -> anyhow::Result<()> {
-    let setup = setup_historic_scanner(
-        Some(0.1),
-        None,
-        BlockNumberOrTag::Earliest,
-        BlockNumberOrTag::Latest,
-    )
-    .await?;
+    let setup =
+        setup_historic_scanner(None, None, BlockNumberOrTag::Earliest, BlockNumberOrTag::Latest)
+            .await?;
     let contract = setup.contract;
     let scanner = setup.scanner;
     let mut stream = setup.stream;

tests/latest_events/basic.rs

@@ -6,7 +6,7 @@ use crate::common::{TestCounter, deploy_counter, setup_common, setup_latest_scan
 use event_scanner::{EventFilter, EventScannerBuilder, assert_closed, assert_next};
 
 #[tokio::test]
-async fn latest_scanner_exact_count_returns_last_events_in_order() -> anyhow::Result<()> {
+async fn exact_count_returns_last_events_in_order() -> anyhow::Result<()> {
     let count = 5;
     let setup = setup_latest_scanner(None, None, count, None, None).await?;
     let contract = setup.contract;
@@ -36,7 +36,7 @@ async fn latest_scanner_exact_count_returns_last_events_in_order() -> anyhow::Re
 }
 
 #[tokio::test]
-async fn latest_scanner_fewer_available_than_count_returns_all() -> anyhow::Result<()> {
+async fn fewer_available_than_count_returns_all() -> anyhow::Result<()> {
     let count = 5;
     let setup = setup_latest_scanner(None, None, count, None, None).await?;
     let contract = setup.contract;
@@ -64,7 +64,7 @@ async fn latest_scanner_fewer_available_than_count_returns_all() -> anyhow::Resu
 }
 
 #[tokio::test]
-async fn latest_scanner_no_events_returns_empty() -> anyhow::Result<()> {
+async fn no_events_returns_empty() -> anyhow::Result<()> {
     let count = 5;
     let setup = setup_latest_scanner(None, None, count, None, None).await?;
     let scanner = setup.scanner;
@@ -81,7 +81,7 @@ async fn latest_scanner_no_events_returns_empty() -> anyhow::Result<()> {
 }
 
 #[tokio::test]
-async fn latest_scanner_respects_range_subset() -> anyhow::Result<()> {
+async fn respects_range_subset() -> anyhow::Result<()> {
     let (_anvil, provider, contract, default_filter) = setup_common(None, None).await?;
     // Mine 6 events, one per tx (auto-mined), then manually mint 2 empty blocks to widen range
     contract.increase().send().await?.watch().await?;
@@ -118,8 +118,7 @@ async fn latest_scanner_respects_range_subset() -> anyhow::Result<()> {
 }
 
 #[tokio::test]
-async fn latest_scanner_multiple_listeners_to_same_event_receive_same_results() -> anyhow::Result<()>
-{
+async fn multiple_listeners_to_same_event_receive_same_results() -> anyhow::Result<()> {
     let count = 5;
     let setup = setup_latest_scanner(None, None, count, None, None).await?;
     let contract = setup.contract;
@@ -161,7 +160,7 @@ async fn latest_scanner_multiple_listeners_to_same_event_receive_same_results()
 }
 
 #[tokio::test]
-async fn latest_scanner_different_filters_receive_different_results() -> anyhow::Result<()> {
+async fn different_filters_receive_different_results() -> anyhow::Result<()> {
     let count = 3;
     let setup = setup_latest_scanner(None, None, count, None, None).await?;
     let contract = setup.contract;
@@ -216,7 +215,7 @@ async fn latest_scanner_different_filters_receive_different_results() -> anyhow:
 }
 
 #[tokio::test]
-async fn latest_scanner_mixed_events_and_filters_return_correct_streams() -> anyhow::Result<()> {
+async fn mixed_events_and_filters_return_correct_streams() -> anyhow::Result<()> {
     let count = 2;
     let setup = setup_latest_scanner(None, None, count, None, None).await?;
     let contract = setup.contract;
@@ -229,17 +228,11 @@ async fn latest_scanner_mixed_events_and_filters_return_correct_streams() -> any
         .event(TestCounter::CountDecreased::SIGNATURE);
     let mut stream_dec = scanner.subscribe(filter_dec);
 
-    // Sequence: inc(1), inc(2), dec(1), inc(2), dec(1)
-    // inc -> 1
-    contract.increase().send().await?.watch().await?;
-    // inc -> 2
-    contract.increase().send().await?.watch().await?;
-    // dec -> 1
-    contract.decrease().send().await?.watch().await?;
-    // inc -> 2
-    contract.increase().send().await?.watch().await?;
-    // dec -> 1
-    contract.decrease().send().await?.watch().await?;
+    contract.increase().send().await?.watch().await?; // inc(1)
+    contract.increase().send().await?.watch().await?; // inc(2)
+    contract.decrease().send().await?.watch().await?; // dec(1)
+    contract.increase().send().await?.watch().await?; // inc(2)
+    contract.decrease().send().await?.watch().await?; // dec(1)
 
     scanner.start().await?;
 
@@ -265,7 +258,7 @@ async fn latest_scanner_mixed_events_and_filters_return_correct_streams() -> any
 }
 
 #[tokio::test]
-async fn latest_scanner_ignores_non_tracked_contract() -> anyhow::Result<()> {
+async fn ignores_non_tracked_contract() -> anyhow::Result<()> {
     // Manual setup to deploy two contracts
     let setup = setup_latest_scanner(None, None, 5, None, None).await?;
     let provider = setup.provider;
@@ -300,7 +293,7 @@ async fn latest_scanner_ignores_non_tracked_contract() -> anyhow::Result<()> {
 }
 
 #[tokio::test]
-async fn latest_scanner_large_gaps_and_empty_ranges() -> anyhow::Result<()> {
+async fn large_gaps_and_empty_ranges() -> anyhow::Result<()> {
     // Manual setup to mine empty blocks
     let (_anvil, provider, contract, default_filter) = setup_common(None, None).await?;
 
@@ -337,7 +330,7 @@ async fn latest_scanner_large_gaps_and_empty_ranges() -> anyhow::Result<()> {
 }
 
 #[tokio::test]
-async fn latest_scanner_boundary_range_single_block() -> anyhow::Result<()> {
+async fn boundary_range_single_block() -> anyhow::Result<()> {
     let (_anvil, provider, contract, default_filter) = setup_common(None, None).await?;
 
     contract.increase().send().await?.watch().await?;