chore(request-response): migrate tests from async-std to tokio

Migrating tests in protocols/request-response from async_std to tokio as per #4449

Pull-Request: #5857.

Author: shiyasmohd

Cargo.lock

@@ -18,7 +18,7 @@ libp2p-core = { workspace = true }
 libp2p-swarm = { workspace = true }
 libp2p-identity = { workspace = true }
 rand = "0.8"
-serde = { version = "1.0", optional = true}
+serde = { version = "1.0", optional = true }
 serde_json = { version = "1.0.117", optional = true }
 smallvec = "1.13.2"
 tracing = { workspace = true }
@@ -30,7 +30,7 @@ cbor = ["dep:serde", "dep:cbor4ii", "libp2p-swarm/macros"]
 
 [dev-dependencies]
 anyhow = "1.0.86"
-async-std = { version = "1.6.2", features = ["attributes"] }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
 rand = "0.8"
 libp2p-swarm-test = { path = "../../swarm-test", features = ["async-std"]}
 futures_ringbuf = "0.4.0"

protocols/request-response/Cargo.toml

@@ -205,7 +205,7 @@ mod tests {
 
     use crate::{cbor::codec::Codec, Codec as _};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn test_codec() {
         let expected_request = TestRequest {
             payload: "test_payload".to_string(),

protocols/request-response/src/cbor.rs

@@ -184,7 +184,7 @@ mod tests {
 
     use crate::Codec;
 
-    #[async_std::test]
+    #[tokio::test]
     async fn test_codec() {
         let expected_request = TestRequest {
             payload: "test_payload".to_string(),

protocols/request-response/src/json.rs

@@ -1,9 +1,8 @@
 use std::{io, iter, pin::pin, time::Duration};
 
 use anyhow::{bail, Result};
-use async_std::task::sleep;
 use async_trait::async_trait;
-use futures::prelude::*;
+use futures::{future::pending, prelude::*};
 use libp2p_identity::PeerId;
 use libp2p_request_response as request_response;
 use libp2p_request_response::ProtocolSupport;
@@ -14,7 +13,7 @@ use request_response::{
 };
 use tracing_subscriber::EnvFilter;
 
-#[async_std::test]
+#[tokio::test]
 async fn report_outbound_failure_on_read_response() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -70,7 +69,7 @@ async fn report_outbound_failure_on_read_response() {
     futures::future::select(server_task, client_task).await;
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn report_outbound_failure_on_write_request() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -113,7 +112,7 @@ async fn report_outbound_failure_on_write_request() {
     futures::future::select(server_task, client_task).await;
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn report_outbound_timeout_on_read_response() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -160,7 +159,7 @@ async fn report_outbound_timeout_on_read_response() {
     futures::future::select(server_task, client_task).await;
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn report_outbound_failure_on_max_streams() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -212,7 +211,7 @@ async fn report_outbound_failure_on_max_streams() {
     futures::future::select(swarm1_task, swarm2_task).await;
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn report_inbound_failure_on_read_request() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -249,7 +248,7 @@ async fn report_inbound_failure_on_read_request() {
     futures::future::select(server_task, client_task).await;
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn report_inbound_failure_on_write_response() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -315,7 +314,7 @@ async fn report_inbound_failure_on_write_response() {
     futures::future::select(server_task, client_task).await;
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn report_inbound_timeout_on_write_response() {
     let _ = tracing_subscriber::fmt()
         .with_env_filter(EnvFilter::from_default_env())
@@ -465,9 +464,10 @@ impl Codec for TestCodec {
 
         match buf[0].try_into()? {
             Action::FailOnReadResponse => Err(io::Error::other("FailOnReadResponse")),
-            Action::TimeoutOnReadResponse => loop {
-                sleep(Duration::MAX).await;
-            },
+            Action::TimeoutOnReadResponse => {
+                pending::<()>().await;
+                Err(io::Error::other("FailOnReadResponse"))
+            }
             action => Ok(action),
         }
     }
@@ -502,9 +502,10 @@ impl Codec for TestCodec {
     {
         match res {
             Action::FailOnWriteResponse => Err(io::Error::other("FailOnWriteResponse")),
-            Action::TimeoutOnWriteResponse => loop {
-                sleep(Duration::MAX).await;
-            },
+            Action::TimeoutOnWriteResponse => {
+                pending::<()>().await;
+                Err(io::Error::other("FailOnWriteResponse"))
+            }
             action => {
                 let bytes = [action.into()];
                 io.write_all(&bytes).await?;

protocols/request-response/tests/error_reporting.rs

@@ -8,7 +8,7 @@ use libp2p_swarm_test::SwarmExt;
 use serde::{Deserialize, Serialize};
 use tracing_subscriber::EnvFilter;
 
-#[async_std::test]
+#[tokio::test]
 #[cfg(feature = "cbor")]
 async fn dial_succeeds_after_adding_peers_address() {
     let _ = tracing_subscriber::fmt()
@@ -34,7 +34,7 @@ async fn dial_succeeds_after_adding_peers_address() {
 
     swarm.dial(peer_id2).unwrap();
 
-    async_std::task::spawn(swarm2.loop_on_next());
+    tokio::spawn(swarm2.loop_on_next());
 
     let (connected_peer_id, connected_address) = swarm
         .wait(|event| match event {

protocols/request-response/tests/peer_address.rs

@@ -32,7 +32,7 @@ use rand::Rng;
 use serde::{Deserialize, Serialize};
 use tracing_subscriber::EnvFilter;
 
-#[async_std::test]
+#[tokio::test]
 #[cfg(feature = "cbor")]
 async fn is_response_outbound() {
     let _ = tracing_subscriber::fmt()
@@ -81,7 +81,7 @@ async fn is_response_outbound() {
 }
 
 /// Exercises a simple ping protocol.
-#[async_std::test]
+#[tokio::test]
 #[cfg(feature = "cbor")]
 async fn ping_protocol() {
     let ping = Ping("ping".to_string().into_bytes());
@@ -173,12 +173,12 @@ async fn ping_protocol() {
         }
     };
 
-    async_std::task::spawn(Box::pin(peer1));
+    tokio::spawn(Box::pin(peer1));
     peer2.await;
 }
 
 /// Exercises a simple ping protocol where peers are not connected prior to request sending.
-#[async_std::test]
+#[tokio::test]
 #[cfg(feature = "cbor")]
 async fn ping_protocol_explicit_address() {
     let ping = Ping("ping".to_string().into_bytes());
@@ -293,11 +293,11 @@ async fn ping_protocol_explicit_address() {
         }
     };
 
-    async_std::task::spawn(Box::pin(peer1));
+    tokio::spawn(Box::pin(peer1));
     peer2.await;
 }
 
-#[async_std::test]
+#[tokio::test]
 #[cfg(feature = "cbor")]
 async fn emits_inbound_connection_closed_failure() {
     let ping = Ping("ping".to_string().into_bytes());
@@ -363,7 +363,7 @@ async fn emits_inbound_connection_closed_failure() {
 /// early close as a protocol violation which results in the connection being closed.
 /// If the substream were not properly closed when dropped, the sender would instead
 /// run into a timeout waiting for the response.
-#[async_std::test]
+#[tokio::test]
 #[cfg(feature = "cbor")]
 async fn emits_inbound_connection_closed_if_channel_is_dropped() {
     let ping = Ping("ping".to_string().into_bytes());
@@ -421,7 +421,7 @@ async fn emits_inbound_connection_closed_if_channel_is_dropped() {
 }
 
 /// Send multiple requests concurrently.
-#[async_std::test]
+#[tokio::test]
 #[cfg(feature = "cbor")]
 async fn concurrent_ping_protocol() {
     use std::{collections::HashMap, num::NonZero};
@@ -537,7 +537,7 @@ async fn concurrent_ping_protocol() {
         assert_eq!(count, num_pings);
     };
 
-    async_std::task::spawn(Box::pin(peer1));
+    tokio::spawn(Box::pin(peer1));
     peer2.await;
 }