feat(network): add geth connection test (#352)

* wip: geth connection test

* set port until disabling discovery sets port

 * ignore test until it is complete

* check that first geth event is SessionEstablished

 * fix endpoint string for the geth instance

* force an incoming connection

 * we should also test establishing a session on an outgoing connections

* use NetworkEventStream helper in connect test

* create outgoing test

* make geth tests use different ports and datadirs

* update geth options

* cargo fmt

* s/geth_socket/reth_socket

* cargo fmt

* fix disc port collisions and re-add incoming test

 * should add test utilities for getting unused disc/p2p ports

Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>

crates/net/network/tests/it/connect.rs

@@ -92,8 +92,6 @@ async fn test_incoming_node_id_blacklist() {
     reth_tracing::init_tracing();
     let secret_key = SecretKey::new(&mut rand::thread_rng());
 
-    let reth_socket = SocketAddr::new([127, 0, 0, 1].into(), 30305);
-
     // instantiate geth and add ourselves as a peer
     let temp_dir = tempfile::tempdir().unwrap().into_path();
     let geth = Geth::new().data_dir(temp_dir).disable_discovery().spawn();
@@ -107,8 +105,11 @@ async fn test_incoming_node_id_blacklist() {
     let ban_list = BanList::new(HashSet::from_iter(vec![geth_peer_id]), HashSet::default());
     let peer_config = PeersConfig::default().with_ban_list(ban_list);
 
+    let reth_p2p_socket = SocketAddr::new([127, 0, 0, 1].into(), 30303);
+    let reth_disc_socket = SocketAddr::new([127, 0, 0, 1].into(), 30304);
     let config = NetworkConfig::builder(Arc::new(TestApi::default()), secret_key)
-        .listener_addr(reth_socket)
+        .listener_addr(reth_p2p_socket)
+        .discovery_addr(reth_disc_socket)
         .peer_config(peer_config)
         .build();
 
@@ -119,7 +120,7 @@ async fn test_incoming_node_id_blacklist() {
 
     // make geth connect to us
     let our_peer_id = handle.peer_id();
-    let our_enode = format!("enode://{}@{}", hex::encode(our_peer_id.0), reth_socket);
+    let our_enode = format!("enode://{}@{}", hex::encode(our_peer_id.0), reth_p2p_socket);
 
     provider.add_peer(our_enode).await.unwrap();
 
@@ -133,3 +134,86 @@ async fn test_incoming_node_id_blacklist() {
     let incoming_peer_id = event_stream.next_session_closed().await.unwrap();
     assert_eq!(incoming_peer_id, geth_peer_id);
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_incoming_connect_with_single_geth() {
+    reth_tracing::init_tracing();
+    let secret_key = SecretKey::new(&mut rand::thread_rng());
+
+    // instantiate geth and add ourselves as a peer
+    let temp_dir = tempfile::tempdir().unwrap().into_path();
+    let geth = Geth::new().data_dir(temp_dir).disable_discovery().spawn();
+    let geth_endpoint = SocketAddr::new([127, 0, 0, 1].into(), geth.port());
+    let provider = Provider::<Http>::try_from(format!("http://{geth_endpoint}")).unwrap();
+
+    // get the peer id we should be expecting
+    let geth_peer_id: PeerId =
+        provider.node_info().await.unwrap().enr.public_key().encode_uncompressed().into();
+
+    let reth_p2p_socket = SocketAddr::new([127, 0, 0, 1].into(), 30305);
+    let reth_disc_socket = SocketAddr::new([127, 0, 0, 1].into(), 30306);
+    let config = NetworkConfig::builder(Arc::new(TestApi::default()), secret_key)
+        .listener_addr(reth_p2p_socket)
+        .discovery_addr(reth_disc_socket)
+        .build();
+
+    let network = NetworkManager::new(config).await.unwrap();
+
+    let handle = network.handle().clone();
+    tokio::task::spawn(network);
+
+    // make geth connect to us
+    let our_peer_id = handle.peer_id();
+    let our_enode = format!("enode://{}@{}", hex::encode(our_peer_id.0), reth_p2p_socket);
+
+    provider.add_peer(our_enode).await.unwrap();
+
+    let events = handle.event_listener();
+    let mut event_stream = NetworkEventStream::new(events);
+
+    // check for a sessionestablished event
+    let incoming_peer_id = event_stream.next_session_established().await.unwrap();
+    assert_eq!(incoming_peer_id, geth_peer_id);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_outgoing_connect_with_single_geth() {
+    reth_tracing::init_tracing();
+    let secret_key = SecretKey::new(&mut rand::thread_rng());
+
+    let reth_p2p_socket = SocketAddr::new([127, 0, 0, 1].into(), 30307);
+    let reth_disc_socket = SocketAddr::new([127, 0, 0, 1].into(), 30308);
+    let config = NetworkConfig::builder(Arc::new(TestApi::default()), secret_key)
+        .listener_addr(reth_p2p_socket)
+        .discovery_addr(reth_disc_socket)
+        .build();
+    let network = NetworkManager::new(config).await.unwrap();
+
+    let handle = network.handle().clone();
+    tokio::task::spawn(network);
+
+    // instantiate geth and add ourselves as a peer
+    let temp_dir = tempfile::tempdir().unwrap().into_path();
+    let geth = Geth::new().disable_discovery().data_dir(temp_dir).spawn();
+
+    let geth_p2p_port = geth.p2p_port().unwrap();
+    let geth_socket = SocketAddr::new([127, 0, 0, 1].into(), geth_p2p_port);
+    let geth_endpoint = SocketAddr::new([127, 0, 0, 1].into(), geth.port()).to_string();
+
+    let provider = Provider::<Http>::try_from(format!("http://{}", geth_endpoint)).unwrap();
+
+    // get the peer id we should be expecting
+    let geth_peer_id: PeerId =
+        provider.node_info().await.unwrap().enr.public_key().encode_uncompressed().into();
+
+    // add geth as a peer then wait for a `SessionEstablished` event
+    handle.add_peer(geth_peer_id, geth_socket);
+
+    // create networkeventstream to get the next session established event easily
+    let events = handle.event_listener();
+    let mut event_stream = NetworkEventStream::new(events);
+
+    // check for a sessionestablished event
+    let incoming_peer_id = event_stream.next_session_established().await.unwrap();
+    assert_eq!(incoming_peer_id, geth_peer_id);
+}