test: refactor

Author: 0xNeshi

tests/latest_events/basic.rs

@@ -17,15 +17,9 @@ async fn latest_scanner_exact_count_returns_last_events_in_order() -> anyhow::Re
     let scanner = setup.scanner;
     let mut stream = setup.stream;
 
-    // Produce 8 events
-    contract.increase().send().await?.watch().await?;
-    contract.increase().send().await?.watch().await?;
-    contract.increase().send().await?.watch().await?;
-    contract.increase().send().await?.watch().await?;
-    contract.increase().send().await?.watch().await?;
-    contract.increase().send().await?.watch().await?;
-    contract.increase().send().await?.watch().await?;
-    contract.increase().send().await?.watch().await?;
+    for _ in 0..8 {
+        contract.increase().send().await?.watch().await?;
+    }
 
     // Ask for the latest 5
     scanner.start().await?;
@@ -278,22 +272,17 @@ async fn latest_scanner_mixed_events_and_filters_return_correct_streams() -> any
 }
 
 #[tokio::test]
-async fn latest_scanner_cross_contract_filtering() -> anyhow::Result<()> {
+async fn latest_scanner_ignores_non_tracked_contract() -> anyhow::Result<()> {
     // Manual setup to deploy two contracts
-    let count = 5;
-    let setup = setup_latest_scanner(None, None, count, None, None).await?;
+    let setup = setup_latest_scanner(None, None, 5, None, None).await?;
     let provider = setup.provider;
-    let mut scanner = setup.scanner;
+    let scanner = setup.scanner;
 
-    let contract_a = deploy_counter(provider.clone()).await?;
-    let contract_b = deploy_counter(provider.clone()).await?;
+    let contract_a = setup.contract;
+    let contract_b = deploy_counter(provider).await?;
 
     // Listener only for contract A CountIncreased
-    let filter_a = EventFilter::new()
-        .contract_address(*contract_a.address())
-        .event(TestCounter::CountIncreased::SIGNATURE);
-
-    let mut stream_a = scanner.subscribe(filter_a);
+    let mut stream_a = setup.stream;
 
     // Emit interleaved events from A and B: A(1), B(1), A(2), B(2), A(3)
     contract_a.increase().send().await?.watch().await?;

tests/live/performance.rs

@@ -1,52 +1,32 @@
-use std::{
-    sync::{
-        Arc,
-        atomic::{AtomicUsize, Ordering},
-    },
-    time::Duration,
-};
+use alloy::primitives::U256;
+use event_scanner::{assert_empty, assert_next};
 
-use event_scanner::Message;
-use tokio::time::timeout;
-use tokio_stream::StreamExt;
-
-use crate::common::{TestCounter, setup_live_scanner};
+use crate::common::{LiveScannerSetup, TestCounter::CountIncreased, setup_live_scanner};
 
 #[tokio::test]
 async fn high_event_volume_no_loss() -> anyhow::Result<()> {
-    let setup = setup_live_scanner(Some(0.05), None, 0).await?;
-    let contract = setup.contract.clone();
-
-    let expected_event_count = 100;
-
-    let scanner = setup.scanner;
-
-    let mut stream = setup.stream.take(expected_event_count);
+    let LiveScannerSetup { contract, provider: _p, scanner, mut stream, anvil: _a } =
+        setup_live_scanner(None, None, 0).await?;
 
     scanner.start().await?;
 
-    for _ in 0..expected_event_count {
-        contract.increase().send().await?.watch().await?;
-    }
-
-    let event_count = Arc::new(AtomicUsize::new(0));
-    let event_count_clone = Arc::clone(&event_count);
-    let event_counting = async move {
-        let mut expected_new_count = 1;
-        while let Some(Message::Data(logs)) = stream.next().await {
-            event_count_clone.fetch_add(logs.len(), Ordering::SeqCst);
-
-            for log in logs {
-                let TestCounter::CountIncreased { newCount } = log.log_decode().unwrap().inner.data;
-                assert_eq!(newCount, expected_new_count);
-                expected_new_count += 1;
-            }
+    tokio::spawn(async move {
+        for _ in 0..100 {
+            contract
+                .increase()
+                .send()
+                .await
+                .expect("should send")
+                .watch()
+                .await
+                .expect("should confirm");
         }
-    };
+    });
 
-    _ = timeout(Duration::from_secs(60), event_counting).await;
-
-    assert_eq!(event_count.load(Ordering::SeqCst), expected_event_count);
+    for new_count in 1..=100 {
+        assert_next!(stream, &[CountIncreased { newCount: U256::from(new_count) }]);
+    }
+    assert_empty!(stream);
 
     Ok(())
 }

tests/sync/from_block.rs

@@ -6,7 +6,7 @@ use alloy::{
 };
 use event_scanner::{ScannerStatus, assert_empty, assert_next};
 
-use crate::common::{TestCounter, setup_sync_scanner};
+use crate::common::{SyncScannerSetup, TestCounter, setup_sync_scanner};
 
 #[tokio::test]
 async fn replays_historical_then_switches_to_live() -> anyhow::Result<()> {
@@ -75,29 +75,37 @@ async fn sync_from_future_block_waits_until_minted() -> anyhow::Result<()> {
     Ok(())
 }
 
-#[test_log::test(tokio::test)]
+#[tokio::test]
 async fn block_confirmations_mitigate_reorgs() -> anyhow::Result<()> {
     // any reorg ≤ 5 should be invisible to consumers
-    let setup = setup_sync_scanner(None, None, BlockNumberOrTag::Earliest, 5).await?;
-    let provider = setup.provider;
-    let contract = setup.contract;
-    let scanner = setup.scanner;
-    let mut stream = setup.stream;
+    let SyncScannerSetup { provider, contract, scanner, mut stream, anvil: _anvil } =
+        setup_sync_scanner(None, None, BlockNumberOrTag::Earliest, 5).await?;
 
     // mine some initial "historic" blocks
-    contract.increase().send().await?.watch().await?;
-    provider.anvil_mine(Some(5), None).await?;
+    for _ in 0..7 {
+        contract.increase().send().await?.watch().await?;
+    }
 
     scanner.start().await?;
 
     // emit "live" events
-    for _ in 0..4 {
+    for _ in 0..2 {
         contract.increase().send().await?.watch().await?;
     }
 
-    // assert only the first events has enough confirmations to be streamed
-    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(1) }]);
+    // assert historic events are streamed in a batch
+    assert_next!(
+        stream,
+        &[
+            TestCounter::CountIncreased { newCount: U256::from(1) },
+            TestCounter::CountIncreased { newCount: U256::from(2) }
+        ]
+    );
+    // switching to "live" phase
     assert_next!(stream, ScannerStatus::SwitchingToLive);
+    // assert confirmed live events are streamed separately
+    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(3) }]);
+    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(4) }]);
     let stream = assert_empty!(stream);
 
     // Perform a shallow reorg on the live tail
@@ -116,13 +124,14 @@ async fn block_confirmations_mitigate_reorgs() -> anyhow::Result<()> {
     provider.anvil_mine(Some(10), None).await?;
 
     // no `ReorgDetected` should be emitted
-    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(2) }]);
-    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(3) }]);
+    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(5) }]);
+    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(6) }]);
+    assert_next!(stream, &[TestCounter::CountIncreased { newCount: U256::from(7) }]);
     assert_next!(
         stream,
         &[
-            TestCounter::CountIncreased { newCount: U256::from(4) },
-            TestCounter::CountIncreased { newCount: U256::from(5) }
+            TestCounter::CountIncreased { newCount: U256::from(8) },
+            TestCounter::CountIncreased { newCount: U256::from(9) }
         ]
     );
     assert_empty!(stream);