Add Sniffer for integration test

`Sniffer` allows intercepting messages exchanged between two roles,
enabling testing whether a message have been sent/received by a
downstream/upstream role.

roles/tests-integration/Cargo.toml

@@ -14,6 +14,7 @@ key-utils = { path = "../../utils/key-utils" }
 minreq = { version = "2.12.0", features = ["https"] }
 once_cell = "1.19.0"
 network_helpers_sv2 = { path = "../roles-utils/network-helpers", features =["with_tokio","with_buffer_pool"] }
+pool_sv2 = { path = "../pool" }
 roles_logic_sv2 = { path = "../../protocols/v2/roles-logic-sv2" }
 tar = "0.4.41"
 tokio = { version="1.36.0",features = ["full","tracing"] }

roles/tests-integration/tests/common/mod.rs

@@ -4,8 +4,8 @@ use bitcoind::{bitcoincore_rpc::RpcApi, BitcoinD, Conf};
 use flate2::read::GzDecoder;
 use key_utils::{Secp256k1PublicKey, Secp256k1SecretKey};
 use once_cell::sync::Lazy;
-use sniffer::Sniffer;
 use pool_sv2::PoolSv2;
+use sniffer::Sniffer;
 use std::{
     collections::HashSet,
     env,
@@ -15,7 +15,6 @@ use std::{
     path::{Path, PathBuf},
     str::FromStr,
     sync::Mutex,
-    time::Duration,
 };
 use tar::Archive;
 
@@ -78,6 +77,12 @@ pub struct TemplateProvider {
     bitcoind: BitcoinD,
 }
 
+impl Drop for TemplateProvider {
+    fn drop(&mut self) {
+        self.stop();
+    }
+}
+
 impl TemplateProvider {
     pub fn start(port: u16) -> Self {
         let path_name = format!("/tmp/.template-provider-{}", port);
@@ -163,11 +168,16 @@ impl TemplateProvider {
     }
 }
 
-pub fn is_port_open(address: SocketAddr) -> bool {
+fn is_port_open(address: SocketAddr) -> bool {
     TcpListener::bind(address).is_err()
 }
 
-pub fn get_available_port() -> u16 {
+pub fn get_available_address() -> SocketAddr {
+    let port = get_available_port();
+    SocketAddr::from(([127, 0, 0, 1], port))
+}
+
+fn get_available_port() -> u16 {
     let mut unique_ports = UNIQUE_PORTS.lock().unwrap();
 
     loop {
@@ -193,9 +203,8 @@ pub async fn start_sniffer(upstream: SocketAddr, downstream: SocketAddr) -> Snif
 }
 
 #[derive(Debug)]
-pub struct TestPoolSv2 {
+struct TestPoolSv2 {
     pub pool: PoolSv2,
-    pub port: u16,
 }
 
 impl TestPoolSv2 {
@@ -205,7 +214,11 @@ impl TestPoolSv2 {
         template_provider_address: Option<SocketAddr>,
     ) -> Self {
         use pool_sv2::mining_pool::{CoinbaseOutput, Configuration};
-        let pool_port = get_available_port();
+        let pool_port = if let Some(listen_addr) = listening_address {
+            listen_addr.port()
+        } else {
+            get_available_port()
+        };
         let listening_address = listening_address
             .unwrap_or(SocketAddr::from_str(&format!("127.0.0.1:{}", pool_port)).unwrap());
         let is_pool_port_open = is_port_open(listening_address);
@@ -238,8 +251,10 @@ impl TestPoolSv2 {
             cert_validity_sec,
             pool_signature,
         );
-        let template_provider_config =
-            pool_sv2::mining_pool::TemplateProviderConfig::new(tp_address, None);
+        let template_provider_config = pool_sv2::mining_pool::TemplateProviderConfig::new(
+            tp_address,
+            Some(authority_public_key),
+        );
         let authority_config =
             pool_sv2::mining_pool::AuthorityConfig::new(authority_public_key, authority_secret_key);
         let config = Configuration::new(
@@ -250,45 +265,31 @@ impl TestPoolSv2 {
         );
         let pool = PoolSv2::new(config);
 
-        Self {
-            pool,
-            port: pool_port,
-        }
+        Self { pool }
     }
 }
 
-pub async fn start_template_provider() -> (TemplateProvider, u16) {
-    let template_provider_port = get_available_port();
-    let template_provider = TemplateProvider::start(template_provider_port);
+pub async fn start_template_provider(tp_port: u16) -> TemplateProvider {
+    let template_provider = TemplateProvider::start(tp_port);
     template_provider.generate_blocks(16);
-    (template_provider, template_provider_port)
+    template_provider
 }
 
-pub async fn start_template_provider_and_pool() -> Result<(PoolSv2, u16, TemplateProvider, u16), ()>
-{
-    let (template_provider, template_provider_port) = start_template_provider().await;
-    let template_provider_address =
-        SocketAddr::from_str(&format!("127.0.0.1:{}", template_provider_port)).unwrap();
-    let test_pool = TestPoolSv2::new(None, None, Some(template_provider_address));
+pub async fn start_pool(
+    listening_address: Option<SocketAddr>,
+    coinbase_outputs: Option<Vec<pool_sv2::mining_pool::CoinbaseOutput>>,
+    template_provider_address: Option<SocketAddr>,
+) -> PoolSv2 {
+    let test_pool = TestPoolSv2::new(
+        listening_address,
+        coinbase_outputs,
+        template_provider_address,
+    );
     let pool = test_pool.pool.clone();
-    let _pool = pool.clone();
+    let pool_clone = pool.clone();
     tokio::task::spawn(async move {
-        assert!(_pool.start().await.is_ok());
+        assert!(pool_clone.start().await.is_ok());
     });
-    // Wait for the pool to start.
-    tokio::time::sleep(Duration::from_secs(1)).await;
-    let pool_listening_address =
-        SocketAddr::from_str(&format!("127.0.0.1:{}", test_pool.port)).unwrap();
-    loop {
-        if is_port_open(pool_listening_address) {
-            break;
-        }
-    }
-    template_provider.stop();
-    Ok((
-        pool,
-        test_pool.port,
-        template_provider,
-        template_provider_port,
-    ))
+    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    pool
 }

roles/tests-integration/tests/pool_integration.rs

@@ -1,31 +1,16 @@
-use std::str::FromStr;
-
 mod common;
 
 #[tokio::test]
 async fn success_pool_template_provider_connection() {
-    assert!(common::start_template_provider_and_pool().await.is_ok());
-}
-
-#[tokio::test]
-async fn pool_bad_coinbase_output() {
-    let (template_provider, template_provider_port) = common::start_template_provider().await;
-    let invalid_coinbase_output = vec![pool_sv2::mining_pool::CoinbaseOutput::new(
-	"P2PK".to_string(),
-	"04466d7fcae563e5cb09a0d1870bb580344804617879a14949cf22285f1bae3f276728176c3c6431f8eeda4538dc37c865e2784f3a9e77d044f33e407797e1278".to_string(),
-  )];
-    let template_provider_address =
-        std::net::SocketAddr::from_str(&format!("127.0.0.1:{}", template_provider_port)).unwrap();
-    let test_pool = common::TestPoolSv2::new(
-        None,
-        Some(invalid_coinbase_output),
-        Some(template_provider_address),
-    );
-    let pool = test_pool.pool.clone();
-    let state = pool.state().await.safe_lock(|s| s.clone()).unwrap();
-    assert_eq!(state, pool_sv2::PoolState::Initial);
-    assert!(pool.start().await.is_err());
-    let state = pool.state().await.safe_lock(|s| s.clone()).unwrap();
-    assert_eq!(state, pool_sv2::PoolState::Initial);
-    template_provider.stop();
+    let sniffer_addr = common::get_available_address();
+    let tp_addr = common::get_available_address();
+    let pool_addr = common::get_available_address();
+    let _tp = common::start_template_provider(tp_addr.port()).await;
+    let sniffer = common::start_sniffer(tp_addr, sniffer_addr).await;
+    let _ = common::start_pool(Some(pool_addr), None, Some(sniffer_addr)).await;
+    assert!(sniffer.expect_downstream_setup_connection());
+    assert!(sniffer.expect_upstream_setup_connection_success());
+    assert!(sniffer.expect_downstream_coinbase_output_data_size());
+    assert!(sniffer.expect_upstream_new_template());
+    assert!(sniffer.expect_upstream_set_new_prev_hash());
 }