Add integration test for mining device auth..

failure with a pool.

This test run a `TemplateProvider`, `PoolSv2` and a `mining_device`
with the auth-failure flag enable which will result in the `PoolSv2` to
reject the `mining_device`.

roles/tests-integration/Cargo.toml

@@ -12,6 +12,10 @@ tar = "0.4.41"
 pool_sv2 = { version = "0.1.0", path = "../pool" }
 key-utils = { version = "1.0.0", path = "../../utils/key-utils" }
 tokio = { version = "1.16.1", features = ["full"] }
+mining_device = { version = "0.1.1", path = "../test-utils/mining-device" }
 
 [lib]
 path = "tests/common/mod.rs"
+
+[features]
+mining_device_reject_auth = ["pool_sv2/MG_reject_auth", "mining_device/abort_mining"]

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

@@ -284,3 +284,13 @@ pub async fn start_template_provider_and_pool() -> Result<(PoolSv2, u16, Templat
         template_provider_port,
     ))
 }
+
+pub struct TestMiningDevice;
+
+impl TestMiningDevice {
+    pub async fn start(pool_address: SocketAddr) -> Result<(), pool_sv2::error::PoolError> {
+        mining_device::connect(pool_address.to_string(), None, None, None, 0)
+            .await
+            .map_err(|e| e.into())
+    }
+}

roles/tests-integration/tests/pool_integration.rs

@@ -29,3 +29,46 @@ async fn pool_bad_coinbase_output() {
     assert_eq!(state, pool_sv2::PoolState::Initial);
     template_provider.stop();
 }
+
+#[cfg(feature = "mining_device_reject_auth")]
+#[tokio::test]
+async fn pool_remove_downstream_after_bad_auth() {
+    let template_provider_port = get_available_port().unwrap();
+    let template_provider = common::TemplateProvider::start(template_provider_port);
+    template_provider.generate_blocks(16);
+    let pool_port = get_available_port().unwrap();
+    let pool_listening_address = SocketAddr::from_str(&format!("127.0.0.1:{}", pool_port)).unwrap();
+    let is_pool_port_open = is_port_open(pool_listening_address);
+    assert_eq!(is_pool_port_open, false);
+    let pool = common::TestPoolSv2::new(
+        pool_listening_address,
+        None,
+        Some(SocketAddr::from_str(&format!("127.0.0.1:{}", template_provider_port)).unwrap()),
+    );
+    let state = pool.state().await.safe_lock(|s| s.clone()).unwrap();
+    assert_eq!(state, pool_sv2::PoolState::Initial);
+    let _pool = pool.clone();
+    tokio::task::spawn(async move {
+        let _ = _pool.start().await;
+    });
+    sleep(Duration::from_secs(1)).await;
+    // Wait for the pool to start.
+    loop {
+        if is_port_open(pool_listening_address) {
+            break;
+        }
+    }
+    let state = pool.state().await.safe_lock(|s| s.clone()).unwrap();
+    assert_eq!(
+        state,
+        pool_sv2::PoolState::Running(pool_sv2::DroppedDownstreams::new())
+    );
+    assert!(common::TestMiningDevice::start(pool_listening_address)
+        .await
+        .is_err());
+    let state = pool.state().await.safe_lock(|s| s.clone()).unwrap();
+    let mut dropped_downstreams = pool_sv2::DroppedDownstreams::new();
+    dropped_downstreams.push(1);
+    assert_eq!(state, pool_sv2::PoolState::Running(dropped_downstreams));
+    template_provider.stop();
+}