feat(p2p/test_utils): add support for app specific test events

Author: CHr15F0x

crates/p2p/src/builder.rs

@@ -132,6 +132,7 @@ impl<B> Builder<B, AppBehaviourSet> {
 impl ApplicationBehaviour for dummy::Behaviour {
     type Command = ();
     type Event = ();
+    type TestEvent = ();
     type State = ();
 
     async fn handle_command(&mut self, _: Self::Command, _: &mut Self::State) {}
@@ -140,6 +141,7 @@ impl ApplicationBehaviour for dummy::Behaviour {
         _: <Self as NetworkBehaviour>::ToSwarm,
         _: &mut Self::State,
         _: mpsc::UnboundedSender<Self::Event>,
+        _: mpsc::UnboundedSender<Self::TestEvent>,
     ) {
     }
     fn domain() -> &'static str {

crates/p2p/src/consensus.rs

@@ -2,7 +2,7 @@
 //! network.
 use std::collections::HashMap;
 
-use libp2p::gossipsub::PublishError;
+use libp2p::gossipsub::{PublishError, TopicHash};
 use libp2p::PeerId;
 use p2p_proto::consensus::{ProposalPart, Vote};
 use pathfinder_common::ContractAddress;
@@ -93,6 +93,17 @@ impl EventKind {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct TestEvent {
+    pub source: PeerId,
+    pub kind: TestEventKind,
+}
+
+#[derive(Debug, Clone)]
+pub enum TestEventKind {
+    Subscribed(TopicHash),
+}
+
 /// The state of the consensus P2P network.
 #[derive(Default, Debug)]
 pub struct State {

crates/p2p/src/consensus/behaviour.rs

@@ -29,6 +29,7 @@ pub struct Behaviour {
 impl ApplicationBehaviour for Behaviour {
     type Command = consensus::Command;
     type Event = consensus::Event;
+    type TestEvent = consensus::TestEvent;
     type State = consensus::State;
 
     #[tracing::instrument(skip(self, state))]
@@ -148,6 +149,7 @@ impl ApplicationBehaviour for Behaviour {
         event: BehaviourEvent,
         state: &mut Self::State,
         event_sender: mpsc::UnboundedSender<Self::Event>,
+        test_event_sender: mpsc::UnboundedSender<Self::TestEvent>,
     ) {
         use gossipsub::Event::*;
         let BehaviourEvent::Gossipsub(e) = event;
@@ -187,6 +189,17 @@ impl ApplicationBehaviour for Behaviour {
                 }
                 _ => {}
             },
+            Subscribed { peer_id, topic } => {
+                tracing::debug!("Peer {} subscribed to topic {}", peer_id, topic);
+
+                #[cfg(test)]
+                {
+                    let _ = test_event_sender.send(consensus::TestEvent {
+                        source: peer_id,
+                        kind: consensus::TestEventKind::Subscribed(topic),
+                    });
+                }
+            }
             _ => {
                 // TODO: Do we care about any other Gossipsub events?
             }

crates/p2p/src/core/tests.rs

@@ -46,14 +46,14 @@ async fn disconnect() {
     peer2.client.disconnect(peer1.peer_id).await.unwrap();
 
     peer1
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionClosed { remote } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
         .await;
 
     peer2
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionClosed { remote } if remote == peer1.peer_id => Some(()),
             _ => None,
         })
@@ -102,17 +102,17 @@ async fn periodic_bootstrap() {
 
     let peer_id2 = peer2.peer_id;
 
-    let peer2_added_to_dht_of_peer1 = peer1.wait_for_test_event(move |e| match e {
+    let peer2_added_to_dht_of_peer1 = peer1.wait_for_core_test_event(move |e| match e {
         TestEvent::PeerAddedToDHT { remote } if remote == peer_id2 => Some(()),
         _ => None,
     });
 
     join(peer2_added_to_dht_of_peer1, async {
         peer2
-            .wait_for_test_event(filter_kademlia_bootstrap_completed)
+            .wait_for_core_test_event(filter_kademlia_bootstrap_completed)
             .await;
         peer2
-            .wait_for_test_event(filter_kademlia_bootstrap_completed)
+            .wait_for_core_test_event(filter_kademlia_bootstrap_completed)
             .await;
     })
     .await;
@@ -155,14 +155,14 @@ async fn reconnect_too_quickly() {
         .unwrap();
 
     peer1
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionEstablished { remote, .. } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
         .await;
 
     peer2
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionEstablished { remote, .. } if remote == peer1.peer_id => Some(()),
             _ => None,
         })
@@ -178,14 +178,14 @@ async fn reconnect_too_quickly() {
     peer1.client.disconnect(peer2.peer_id).await.unwrap();
 
     peer1
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionClosed { remote } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
         .await;
 
     peer2
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionClosed { remote } if remote == peer1.peer_id => Some(()),
             _ => None,
         })
@@ -202,14 +202,14 @@ async fn reconnect_too_quickly() {
 
     // The connection is established.
     peer1
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionEstablished { remote, .. } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
         .await;
 
     peer2
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionEstablished { remote, .. } if remote == peer1.peer_id => Some(()),
             _ => None,
         })
@@ -241,14 +241,14 @@ async fn duplicate_connection() {
         .unwrap();
 
     peer1
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionEstablished { remote, .. } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
         .await;
 
     peer2
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionEstablished { remote, .. } if remote == peer1.peer_id => Some(()),
             _ => None,
         })
@@ -267,14 +267,14 @@ async fn duplicate_connection() {
         .unwrap();
 
     peer1_copy
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionEstablished { remote, .. } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
         .await;
 
     peer1_copy
-        .wait_for_test_event(move |e| match e {
+        .wait_for_core_test_event(move |e| match e {
             TestEvent::ConnectionClosed { remote, .. } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
@@ -364,7 +364,7 @@ async fn outbound_peer_eviction() {
     assert!(peers.contains_key(&inbound1.peer_id));
 
     // Ensure that outbound1 actually got disconnected.
-    peer.wait_for_test_event(move |e| match e {
+    peer.wait_for_core_test_event(move |e| match e {
         TestEvent::ConnectionClosed { remote, .. } if remote == outbound1.peer_id => Some(()),
         _ => None,
     })
@@ -443,7 +443,7 @@ async fn inbound_peer_eviction() {
 
     // Ensure that a peer got disconnected.
     let disconnected = peer
-        .wait_for_test_event(|e| match e {
+        .wait_for_core_test_event(|e| match e {
             TestEvent::ConnectionClosed { remote, .. }
                 if inbound_peers.iter().take(25).any(|p| p.peer_id == remote) =>
             {
@@ -496,7 +496,7 @@ async fn evicted_peer_reconnection() {
 
     // Check that peer2 got evicted.
     peer1
-        .wait_for_test_event(|e| match e {
+        .wait_for_core_test_event(|e| match e {
             TestEvent::ConnectionClosed { remote, .. } if remote == peer2.peer_id => Some(()),
             _ => None,
         })
@@ -517,7 +517,7 @@ async fn evicted_peer_reconnection() {
         .await
         .unwrap();
     peer2
-        .wait_for_test_event(|e| match e {
+        .wait_for_core_test_event(|e| match e {
             TestEvent::ConnectionClosed { remote, .. } if remote == peer1.peer_id => Some(()),
             _ => None,
         })
@@ -529,7 +529,7 @@ async fn evicted_peer_reconnection() {
 
     // peer3 gets evicted.
     peer1
-        .wait_for_test_event(|e| match e {
+        .wait_for_core_test_event(|e| match e {
             TestEvent::ConnectionClosed { remote, .. } if remote == peer3.peer_id => Some(()),
             _ => None,
         })

crates/p2p/src/lib.rs

@@ -88,8 +88,9 @@ mod builder_phase {
 /// interacts with the network:
 /// - Commands: Actions requested by the application to be executed by the
 ///   network
-/// - Events: Notifications from the network that the application needs to
-///   handle
+/// - Events: Application-specific notifications from the network that the
+///   application needs to handle
+/// - TestEvents: Application-specific events that are only emitted in tests
 /// - State: Data needed to track ongoing operations
 ///
 /// This trait is implemented by application-specific network behaviors (like
@@ -99,6 +100,9 @@ pub trait ApplicationBehaviour: NetworkBehaviour {
     type Command: std::fmt::Debug;
     /// The type of events that the p2p network can emit to the outside world.
     type Event;
+    /// The type of events that the p2p network can emit to the outside world in
+    /// tests only.
+    type TestEvent;
     /// State needed to track pending network operations and their responses.
     type State;
 
@@ -115,6 +119,7 @@ pub trait ApplicationBehaviour: NetworkBehaviour {
         event: <Self as NetworkBehaviour>::ToSwarm,
         state: &mut Self::State,
         event_sender: mpsc::UnboundedSender<Self::Event>,
+        test_event_sender: mpsc::UnboundedSender<Self::TestEvent>,
     ) -> impl Future<Output = ()> + Send;
 
     /// Returns the domain string used for marker file creation in integration