ref: move tests to robust sub

Author: LeoPatOZ

src/robust_provider/provider.rs

@@ -318,15 +318,13 @@ impl<N: Network> RobustProvider<N> {
 mod tests {
     use super::*;
     use crate::robust_provider::{
-        RobustProviderBuilder,
-        builder::DEFAULT_SUBSCRIPTION_TIMEOUT,
-        subscription::{DEFAULT_RECONNECT_INTERVAL, RobustSubscriptionStream},
+        RobustProviderBuilder, builder::DEFAULT_SUBSCRIPTION_TIMEOUT,
+        subscription::DEFAULT_RECONNECT_INTERVAL,
     };
-    use alloy::providers::{ProviderBuilder, WsConnect, ext::AnvilApi};
-    use alloy_node_bindings::{Anvil, AnvilInstance};
+    use alloy::providers::{ProviderBuilder, WsConnect};
+    use alloy_node_bindings::Anvil;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use tokio::time::sleep;
-    use tokio_stream::StreamExt;
 
     fn test_provider(timeout: u64, max_retries: usize, min_delay: u64) -> RobustProvider {
         RobustProvider {
@@ -340,103 +338,6 @@ mod tests {
         }
     }
 
-    const SHORT_TIMEOUT: Duration = Duration::from_millis(300);
-    const RECONNECT_INTERVAL: Duration = Duration::from_millis(500);
-    const BUFFER_TIME: Duration = Duration::from_millis(100);
-
-    async fn spawn_ws_anvil() -> anyhow::Result<(AnvilInstance, RootProvider)> {
-        let anvil = Anvil::new().try_spawn()?;
-        let provider = ProviderBuilder::new().connect(anvil.ws_endpoint_url().as_str()).await?;
-        Ok((anvil, provider.root().to_owned()))
-    }
-
-    fn assert_backend_gone_or_timeout(err: Error) {
-        match err {
-            Error::Timeout => {}
-            Error::RpcError(e) => {
-                assert!(
-                    matches!(e.as_ref(), RpcError::Transport(TransportErrorKind::BackendGone)),
-                    "Expected BackendGone error, got: {e:?}",
-                );
-            }
-            Error::BlockNotFound(_) => {
-                panic!("Unexpected BlockNotFound error");
-            }
-        }
-    }
-
-    #[macro_export]
-    macro_rules! assert_stream_finished {
-        ($stream: expr) => {
-            $crate::assert_stream_finished!($stream, finish_secs = 3)
-        };
-        ($stream: expr, finish_secs = $finish: expr) => {{
-            let next_item = tokio_stream::StreamExt::next(&mut $stream).await;
-            match next_item {
-                Some(Ok(item)) => panic!("Expected no item during quiet window, got: {:?}", item),
-                None => {}
-                Some(Err(e)) => {
-                    assert!(matches!(e, Error::Timeout), "Expected Timeout error, got: {:?}", e);
-
-                    let second = tokio::time::timeout(
-                        std::time::Duration::from_secs($finish),
-                        tokio_stream::StreamExt::next(&mut $stream),
-                    )
-                    .await
-                    .expect("expected stream to finish after quiet window");
-                    assert!(second.is_none(), "Expected stream to be finished, got: {:?}", second);
-                }
-            }
-        }};
-    }
-
-    #[macro_export]
-    macro_rules! assert_next_block {
-        ($stream: expr, $expected: expr) => {
-            assert_next_block!($stream, $expected, timeout = 5)
-        };
-        ($stream: expr, $expected: expr, timeout = $secs: expr) => {
-            let message = tokio::time::timeout(
-                std::time::Duration::from_secs($secs),
-                tokio_stream::StreamExt::next(&mut $stream),
-            )
-            .await
-            .expect("timed out");
-            if let Some(block) = message {
-                match block {
-                    Ok(block) => assert_eq!(block.number, $expected),
-                    Err(e) => panic!("Got err {e:?}"),
-                }
-            } else {
-                panic!("Expected block {:?}, got: {message:?}", $expected)
-            }
-        };
-    }
-
-    /// Waits for current provider to timeout, then mines on `next_provider` to trigger failover.
-    async fn trigger_failover_with_delay(
-        stream: &mut RobustSubscriptionStream<alloy::network::Ethereum>,
-        next_provider: RootProvider,
-        expected_block: u64,
-        extra_delay: Duration,
-    ) -> anyhow::Result<()> {
-        let task = tokio::spawn(async move {
-            sleep(SHORT_TIMEOUT + extra_delay + BUFFER_TIME).await;
-            next_provider.anvil_mine(Some(1), None).await.unwrap();
-        });
-        assert_next_block!(*stream, expected_block);
-        task.await?;
-        Ok(())
-    }
-
-    async fn trigger_failover(
-        stream: &mut RobustSubscriptionStream<alloy::network::Ethereum>,
-        next_provider: RootProvider,
-        expected_block: u64,
-    ) -> anyhow::Result<()> {
-        trigger_failover_with_delay(stream, next_provider, expected_block, Duration::ZERO).await
-    }
-
     #[tokio::test]
     async fn test_retry_with_timeout_succeeds_on_first_attempt() {
         let provider = test_provider(100, 3, 10);
@@ -567,176 +468,4 @@ mod tests {
 
         Ok(())
     }
-
-    #[tokio::test]
-    async fn ws_fails_http_fallback_returns_primary_error() -> anyhow::Result<()> {
-        // Setup: Create WS primary and HTTP fallback
-        let anvil_1 = Anvil::new().try_spawn()?;
-        let ws_provider =
-            ProviderBuilder::new().connect(anvil_1.ws_endpoint_url().as_str()).await?;
-
-        let anvil_2 = Anvil::new().try_spawn()?;
-        let http_provider = ProviderBuilder::new().connect_http(anvil_2.endpoint_url());
-
-        let robust = RobustProviderBuilder::fragile(ws_provider.clone())
-            .fallback(http_provider.clone())
-            .subscription_timeout(Duration::from_secs(1))
-            .build()
-            .await?;
-
-        // Test: Verify subscription works on primary
-        let subscription = robust.subscribe_blocks().await?;
-        let mut stream = subscription.into_stream();
-
-        ws_provider.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 1);
-
-        ws_provider.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 2);
-
-        // Verify: HTTP fallback can't provide subscription, so we get an error
-        let task = tokio::spawn(async move {
-            sleep(SHORT_TIMEOUT + BUFFER_TIME).await;
-            http_provider.anvil_mine(Some(1), None).await.unwrap();
-        });
-        let err = stream.next().await.unwrap().unwrap_err();
-        task.await?;
-        assert_backend_gone_or_timeout(err);
-
-        let next = stream.next().await;
-        assert!(next.is_none(), "Expected stream to be finished, got: {next:?}");
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn robust_subscription_stream_with_failover() -> anyhow::Result<()> {
-        let (_anvil_1, primary) = spawn_ws_anvil().await?;
-        let (_anvil_2, fallback) = spawn_ws_anvil().await?;
-
-        let robust = RobustProviderBuilder::fragile(primary.clone())
-            .fallback(fallback.clone())
-            .subscription_timeout(SHORT_TIMEOUT)
-            .build()
-            .await?;
-
-        let subscription = robust.subscribe_blocks().await?;
-        let mut stream = subscription.into_stream();
-
-        // Test: Primary works initially
-        primary.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 1);
-
-        primary.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 2);
-
-        // After timeout, should failover to fallback provider
-        trigger_failover(&mut stream, fallback.clone(), 1).await?;
-
-        fallback.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 2);
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn subscription_reconnects_to_primary() -> anyhow::Result<()> {
-        let (_anvil_1, primary) = spawn_ws_anvil().await?;
-        let (_anvil_2, fallback) = spawn_ws_anvil().await?;
-
-        let robust = RobustProviderBuilder::fragile(primary.clone())
-            .fallback(fallback.clone())
-            .subscription_timeout(SHORT_TIMEOUT)
-            .reconnect_interval(RECONNECT_INTERVAL)
-            .build()
-            .await?;
-
-        let subscription = robust.subscribe_blocks().await?;
-        let mut stream = subscription.into_stream();
-
-        // Start on primary
-        primary.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 1);
-
-        // PP times out -> FP1
-        trigger_failover(&mut stream, fallback.clone(), 1).await?;
-
-        fallback.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 2);
-
-        // FP1 times out -> PP (reconnect succeeds)
-        trigger_failover(&mut stream, primary.clone(), 2).await?;
-
-        // PP times out -> FP1 (fallback index was reset)
-        trigger_failover(&mut stream, fallback.clone(), 3).await?;
-
-        fallback.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 4);
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn subscription_cycles_through_multiple_fallbacks() -> anyhow::Result<()> {
-        let (anvil_pp, primary) = spawn_ws_anvil().await?;
-        let (_anvil_1, fb_1) = spawn_ws_anvil().await?;
-        let (_anvil_2, fb_2) = spawn_ws_anvil().await?;
-
-        let robust = RobustProviderBuilder::fragile(primary.clone())
-            .fallback(fb_1.clone())
-            .fallback(fb_2.clone())
-            .subscription_timeout(SHORT_TIMEOUT)
-            .call_timeout(SHORT_TIMEOUT)
-            .build()
-            .await?;
-
-        let subscription = robust.subscribe_blocks().await?;
-        let mut stream = subscription.into_stream();
-
-        // Start on primary
-        primary.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 1);
-
-        // Kill primary - all future PP reconnection attempts will fail
-        drop(anvil_pp);
-
-        // PP times out -> FP1
-        trigger_failover(&mut stream, fb_1.clone(), 1).await?;
-
-        // FP1 times out -> tries PP (fails, takes call_timeout) -> FP2
-        trigger_failover_with_delay(&mut stream, fb_2.clone(), 1, SHORT_TIMEOUT).await?;
-
-        fb_2.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 2);
-
-        // FP2 times out -> tries PP (fails) -> no more fallbacks -> error
-        sleep(SHORT_TIMEOUT * 2 + BUFFER_TIME).await;
-        let err = stream.next().await.unwrap().unwrap_err();
-        assert_backend_gone_or_timeout(err);
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn subscription_fails_with_no_fallbacks() -> anyhow::Result<()> {
-        let (_anvil, provider) = spawn_ws_anvil().await?;
-
-        let robust = RobustProviderBuilder::fragile(provider.clone())
-            .subscription_timeout(SHORT_TIMEOUT)
-            .build()
-            .await?;
-
-        let subscription = robust.subscribe_blocks().await?;
-        let mut stream = subscription.into_stream();
-
-        provider.anvil_mine(Some(1), None).await?;
-        assert_next_block!(stream, 1);
-
-        // No fallback available - should error after timeout
-        sleep(SHORT_TIMEOUT + BUFFER_TIME).await;
-        let err = stream.next().await.unwrap().unwrap_err();
-        assert_backend_gone_or_timeout(err);
-
-        Ok(())
-    }
 }