test(p2p): tests for inital peers connections

akoptelov · akoptelov · commit 0edfed12aad9 · 2023-12-22T17:35:42.000+03:00
diff --git a/node/testing/src/scenarios/driver.rs b/node/testing/src/scenarios/driver.rs
@@ -258,6 +258,20 @@ impl<'cluster> Driver<'cluster> {
         (node_id, peer_id)
     }
 
+    pub fn add_rust_node_with<Item, F>(
+        &mut self,
+        testing_config: RustNodeTestingConfig,
+        mut f: F,
+    ) -> (ClusterNodeId, Item)
+    where
+        F: FnMut(&State) -> Item,
+    {
+        let node_id = self.runner.add_rust_node(testing_config);
+        let state = self.runner.node(node_id).unwrap().state();
+        let item = f(state);
+        (node_id, item)
+    }
+
     pub fn inner(&self) -> &ClusterRunner {
         &self.runner
     }
@@ -293,7 +307,6 @@ pub async fn wait_for_nodes_listening_on_localhost<'cluster>(
     driver.run_until(duration, pred).await
 }
 
-
 /// Creates `num` Rust nodes in the cluster
 pub fn add_rust_nodes<'cluster, N, NodeIds, PeerIds>(
     driver: &mut Driver,
@@ -310,3 +323,22 @@ where
         .map(|_| driver.add_rust_node(config.clone()))
         .unzip()
 }
+
+/// Creates `num` Rust nodes in the cluster
+pub fn add_rust_nodes_with<'cluster, N, NodeIds, Items, Item, F>(
+    driver: &mut Driver,
+    num: N,
+    config: RustNodeTestingConfig,
+    mut f: F,
+) -> (NodeIds, Items)
+where
+    N: Into<u16>,
+    NodeIds: Default + Extend<ClusterNodeId>,
+    Items: Default + Extend<Item>,
+    F: FnMut(&State) -> Item,
+{
+    (0..num.into())
+        .into_iter()
+        .map(|_| driver.add_rust_node_with(config.clone(), &mut f))
+        .unzip()
+}
diff --git a/node/testing/src/scenarios/p2p/basic_outgoing_connections.rs b/node/testing/src/scenarios/p2p/basic_outgoing_connections.rs
@@ -6,8 +6,9 @@ use crate::{
     cluster::ClusterNodeId,
     node::RustNodeTestingConfig,
     scenarios::{
-        add_rust_nodes, as_connection_finalized_event, connection_finalized_event,
-        wait_for_nodes_listening_on_localhost, ClusterRunner, Driver, connection_finalized_with_res_event,
+        add_rust_nodes, add_rust_nodes_with, as_connection_finalized_event,
+        connection_finalized_event, connection_finalized_with_res_event,
+        wait_for_nodes_listening_on_localhost, ClusterRunner, Driver,
     },
 };
 
@@ -142,7 +143,7 @@ impl DontConnectToNodeWithSameId {
         let (node, _) = driver.add_rust_node(
             RustNodeTestingConfig::berkeley_default()
                 .libp2p_port(port)
-                .with_peer_id(bytes)
+                .with_peer_id(bytes),
         );
         // wait for it to be ready
         assert!(
@@ -156,7 +157,7 @@ impl DontConnectToNodeWithSameId {
         let (node_ut, _) = driver.add_rust_node(
             RustNodeTestingConfig::berkeley_default()
                 .libp2p_port(node_ut_port)
-                .with_peer_id(bytes)
+                .with_peer_id(bytes),
         );
 
         driver
@@ -175,7 +176,10 @@ impl DontConnectToNodeWithSameId {
             .await
             .unwrap();
 
-        assert!(connected.is_none(), "the node sholdn't try to connect to itself");
+        assert!(
+            connected.is_none(),
+            "the node sholdn't try to connect to itself"
+        );
     }
 }
 
@@ -189,7 +193,10 @@ impl DontConnectToSelfInitialPeer {
 
         let port = 11109;
         let bytes: [u8; 32] = rand::random();
-        let peer_id = SecretKey::from_bytes(bytes.clone()).public_key().peer_id().to_libp2p_string();
+        let peer_id = SecretKey::from_bytes(bytes.clone())
+            .public_key()
+            .peer_id()
+            .to_libp2p_string();
         let self_maddr = format!("/ip4/127.0.0.1/tcp/{port}/p2p/{peer_id}")
             .parse()
             .unwrap();
@@ -214,7 +221,10 @@ impl DontConnectToSelfInitialPeer {
             .await
             .unwrap();
 
-        assert!(connected.is_none(), "the node sholdn't try to connect to itself");
+        assert!(
+            connected.is_none(),
+            "the node sholdn't try to connect to itself"
+        );
     }
 }
 
@@ -229,8 +239,10 @@ impl DontConnectToInitialPeerWithSameId {
         let port = 11108;
         let node_ut_port = 11109;
         let bytes: [u8; 32] = rand::random();
-        let peer_id = SecretKey::from_bytes(bytes.clone()).public_key().peer_id().to_libp2p_string();
-
+        let peer_id = SecretKey::from_bytes(bytes.clone())
+            .public_key()
+            .peer_id()
+            .to_libp2p_string();
 
         let self_maddr = format!("/ip4/127.0.0.1/tcp/{port}/p2p/{peer_id}")
             .parse()
@@ -240,7 +252,7 @@ impl DontConnectToInitialPeerWithSameId {
         let (node, _) = driver.add_rust_node(
             RustNodeTestingConfig::berkeley_default()
                 .libp2p_port(port)
-                .with_peer_id(bytes)
+                .with_peer_id(bytes),
         );
         // wait for it to be ready
         assert!(
@@ -266,6 +278,143 @@ impl DontConnectToInitialPeerWithSameId {
             .await
             .unwrap();
 
-        assert!(connected.is_none(), "the node sholdn't try to connect to itself");
+        assert!(
+            connected.is_none(),
+            "the node sholdn't try to connect to itself"
+        );
+    }
+}
+
+/// Node should be able to connect to all initial peers.
+#[derive(documented::Documented, Default, Clone, Copy)]
+pub struct ConnectToInitialPeers;
+
+impl ConnectToInitialPeers {
+    pub async fn run<'cluster>(self, runner: ClusterRunner<'cluster>) {
+        const MAX: u8 = 32;
+
+        let mut driver = Driver::new(runner);
+
+        let (peers, port_peer_ids): (Vec<ClusterNodeId>, Vec<_>) = add_rust_nodes_with(
+            &mut driver,
+            MAX,
+            RustNodeTestingConfig::berkeley_default(),
+            |state| {
+                let config = &state.p2p.config;
+                let port = config.libp2p_port.unwrap();
+                let peer_id = config.identity_pub_key.peer_id();
+                (port, peer_id)
+            },
+        );
+
+        // wait for all peers to listen
+        let satisfied = wait_for_nodes_listening_on_localhost(
+            &mut driver,
+            Duration::from_secs(3 * 60),
+            peers.clone(),
+        )
+        .await
+        .unwrap();
+        assert!(satisfied, "all peers should be listening");
+
+        let initial_peers = port_peer_ids
+            .iter()
+            .map(|(port, peer_id)| {
+                format!(
+                    "/ip4/127.0.0.1/tcp/{port}/p2p/{peer_id}",
+                    peer_id = peer_id.clone().to_libp2p_string()
+                )
+                .parse()
+                .unwrap()
+            })
+            .collect::<Vec<_>>();
+        let (node_ut, _) = driver
+            .add_rust_node(RustNodeTestingConfig::berkeley_default().initial_peers(initial_peers));
+
+        // matches event "the node established connection with peer"
+        let mut peer_ids =
+            BTreeSet::from_iter(port_peer_ids.into_iter().map(|(_, peer_id)| peer_id));
+        let pred = |node_id, event: &_, _state: &_| {
+            if node_id != node_ut {
+                false
+            } else if let Some((peer_id, res)) = as_connection_finalized_event(event) {
+                assert!(res.is_ok(), "connection to {peer_id} should succeed");
+                peer_ids.remove(&peer_id);
+                peer_ids.is_empty()
+            } else {
+                false
+            }
+        };
+
+        let satisfied = driver
+            .run_until(Duration::from_secs(3 * 60), pred)
+            .await
+            .unwrap();
+        assert!(satisfied, "did not connect to peers: {:?}", peer_ids);
+    }
+}
+
+/// Node should be able to connect to all initial peers after they become ready.
+#[derive(documented::Documented, Default, Clone, Copy)]
+pub struct ConnectToInitialPeersBecomeReady;
+
+impl ConnectToInitialPeersBecomeReady {
+    pub async fn run<'cluster>(self, runner: ClusterRunner<'cluster>) {
+        const MAX: u16 = 32;
+
+        let mut driver = Driver::new(runner);
+
+        let port_bytes: Vec<_> = (0..MAX)
+            .into_iter()
+            .map(|p| (12000 + p, rand::random::<[u8; 32]>()))
+            .collect();
+
+        let initial_peers = port_bytes
+            .iter()
+            .map(|(port, bytes)| {
+                let secret_key = SecretKey::from_bytes(bytes.clone());
+                format!(
+                    "/ip4/127.0.0.1/tcp/{port}/p2p/{peer_id}",
+                    peer_id = secret_key.public_key().peer_id().to_libp2p_string()
+                )
+                .parse()
+                .unwrap()
+            })
+            .collect::<Vec<_>>();
+        let (node_ut, _) = driver
+            .add_rust_node(RustNodeTestingConfig::berkeley_default().initial_peers(initial_peers));
+
+        driver.wait_for(Duration::from_secs(10), |_, _, _| false).await.unwrap();
+
+        let (_peers, mut peer_ids): (Vec<ClusterNodeId>, BTreeSet<PeerId>) = port_bytes
+            .into_iter()
+            .map(|(port, bytes)| {
+                driver.add_rust_node(
+                    RustNodeTestingConfig::berkeley_default()
+                        .libp2p_port(port)
+                        .with_peer_id(bytes),
+                )
+            })
+            .unzip();
+
+        // matches event "the node established connection with peer"
+        let pred = |node_id, event: &_, _state: &_| {
+            if node_id != node_ut {
+                false
+            } else if let Some((peer_id, res)) = as_connection_finalized_event(event) {
+                if res.is_ok() {
+                    peer_ids.remove(&peer_id);
+                }
+                peer_ids.is_empty()
+            } else {
+                false
+            }
+        };
+
+        let satisfied = driver
+            .run_until(Duration::from_secs(3 * 60), pred)
+            .await
+            .unwrap();
+        assert!(satisfied, "did not connect to peers: {:?}", peer_ids);
     }
 }
diff --git a/node/testing/tests/p2p_basic_outgoing.rs b/node/testing/tests/p2p_basic_outgoing.rs
@@ -1,6 +1,7 @@
 use openmina_node_testing::scenarios::p2p::basic_outgoing_connections::{
-    DontConnectToInitialPeerWithSameId, DontConnectToNodeWithSameId, DontConnectToSelfInitialPeer,
-    MakeMultipleOutgoingConnections, MakeOutgoingConnection,
+    ConnectToInitialPeers, ConnectToInitialPeersBecomeReady, DontConnectToInitialPeerWithSameId,
+    DontConnectToNodeWithSameId, DontConnectToSelfInitialPeer, MakeMultipleOutgoingConnections,
+    MakeOutgoingConnection,
 };
 
 mod common;
@@ -15,6 +16,7 @@ scenario_test!(
     MakeMultipleOutgoingConnections,
     MakeMultipleOutgoingConnections
 );
+
 scenario_test!(
     dont_connect_to_node_same_id,
     DontConnectToNodeWithSameId,
@@ -30,3 +32,14 @@ scenario_test!(
     DontConnectToSelfInitialPeer,
     DontConnectToSelfInitialPeer
 );
+
+scenario_test!(
+    connect_to_all_initial_peers,
+    ConnectToInitialPeers,
+    ConnectToInitialPeers
+);
+scenario_test!(
+    connect_to_all_initial_peers_become_ready,
+    ConnectToInitialPeersBecomeReady,
+    ConnectToInitialPeersBecomeReady
+);
diff --git a/p2p/testing.md b/p2p/testing.md
@@ -59,6 +59,7 @@ The node can obtain its address from other peers. It shouldn't use it when conne
 
 TODO: what if the number of initial peers exceeds the max number of peers?
 
+- [`p2p_basic_outgoing(connect_to_all_initial_peers)`](../node/testing/src/scenarios/p2p/basic_outgoing_connections.rs#L293)
 - [multi_node_initial_joining](../node/testing/src/scenarios/multi_node/basic_connectivity_initial_joining.rs) (partially?)
 
 ### Node should be able to connect to initial peers eventually, even if initially they are not available.
@@ -69,7 +70,8 @@ below).
 
 TODO: Use cases where this is important.
 
-**Tests:** TODO
+**Tests:**
+- [`p2p_basic_outgoing(connect_to_all_initial_peers_become_ready)`](../node/testing/src/scenarios/p2p/basic_outgoing_connections.rs#L362)
 
 ### Node should have a reasonable retry rate for reconnection
 
