Setup for integration tests

Aims to provide common utility functions that can run the different
roles in an integration test enviornement.

.github/workflows/test.yaml

@@ -40,6 +40,10 @@ jobs:
           cargo build --manifest-path=roles/Cargo.toml
           cargo build --manifest-path=utils/Cargo.toml
 
+      - name: Integration tests
+        run: |
+          cargo test --manifest-path=roles/Cargo.toml --verbose --test '*' -- --nocapture
+
       - name: Run sv1-client-and-server example
         run: |
           cargo run --manifest-path=examples/sv1-client-and-server/Cargo.toml --bin client_and_server -- 60

roles/Cargo.toml

@@ -23,6 +23,7 @@ members = [
     "translator",
     "jd-client",
     "jd-server",
+    "tests-integration",
 ]
 
 [profile.dev]

roles/tests-integration/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "tests-integration"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tokio = "1.38.1"
+
+[dev-dependencies]
+jd_client = { version = "0.1.1", path = "../jd-client" }
+translator_sv2 = { version = "0.1.0", path = "../translator" }
+jd_server = { version = "0.1.1", path = "../jd-server" }
+pool_sv2 = { version = "0.1.1", path = "../pool" }
+async-channel = "1.5.1"
+async-std = { version = "1.8.0" }
+key-utils = { version = "1.1.0", path = "../../utils/key-utils" }
+
+[[test]]
+name = "integration_e2e"
+
+

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

@@ -0,0 +1,194 @@
+use std::sync::Arc;
+
+use async_std::net::TcpStream;
+use key_utils::{Secp256k1PublicKey, Secp256k1SecretKey};
+
+pub async fn start_job_declarator_server() {
+    let authority_public_key = Secp256k1PublicKey::try_from(
+        "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72".to_string(),
+    )
+    .expect("failed");
+    let authority_secret_key = Secp256k1SecretKey::try_from(
+        "mkDLTBBRxdBv998612qipDYoTK3YUrqLe8uWw7gu3iXbSrn2n".to_string(),
+    )
+    .expect("failed");
+    let cert_validity_sec = 3600;
+    let listen_jd_address = "0.0.0.0:34264".to_string();
+    let core_rpc_url = "http://75.119.150.111".to_string();
+    let core_rpc_port = 48332;
+    let core_rpc_user = "username".to_string();
+    let core_rpc_pass = "password".to_string();
+    let coinbase_outputs = vec![jd_server::CoinbaseOutput::new(
+        "P2WPKH".to_string(),
+        "036adc3bdf21e6f9a0f0fb0066bf517e5b7909ed1563d6958a10993849a7554075".to_string(),
+    )];
+    let config = jd_server::Configuration::new(
+        listen_jd_address,
+        authority_public_key,
+        authority_secret_key,
+        cert_validity_sec,
+        coinbase_outputs,
+        core_rpc_url,
+        core_rpc_port,
+        core_rpc_user,
+        core_rpc_pass,
+        std::time::Duration::from_secs(5),
+    );
+    jd_server::start_jd_server(config);
+}
+
+pub async fn start_job_declarator_client(
+) -> (jd_client::JobDeclaratorClient, jd_client::ProxyConfig) {
+    let downstream_address = "127.0.0.1".to_string();
+    let downstream_port = 34265;
+    let max_supported_version = 2;
+    let min_supported_version = 2;
+    let min_extranonce2_size = 8;
+    let withhold = false;
+    let authority_pubkey = Secp256k1PublicKey::try_from(
+        "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72".to_string(),
+    )
+    .expect("failed");
+    let authority_secret_key = Secp256k1SecretKey::try_from(
+        "mkDLTBBRxdBv998612qipDYoTK3YUrqLe8uWw7gu3iXbSrn2n".to_string(),
+    )
+    .expect("failed");
+    let cert_validity_sec = 3600;
+    let retry = 10;
+    let tp_address = "75.119.150.111:8442".to_string();
+    let tp_authority_public_key = Secp256k1PublicKey::try_from(
+        "9azQdassggC7L3YMVcZyRJmK7qrFDj5MZNHb4LkaUrJRUhct92W".to_string(),
+    )
+    .expect("failed");
+    let pool_address = "127.0.0.1:34254".to_string();
+    let jd_address = "127.0.0.1:34264".to_string();
+    let pool_signature = "Stratum v2 SRI Pool".to_string();
+    use jd_client::proxy_config::CoinbaseOutput;
+    let coinbase_outputs: CoinbaseOutput = CoinbaseOutput::new(
+        "P2WPKH".to_string(),
+        "036adc3bdf21e6f9a0f0fb0066bf517e5b7909ed1563d6958a10993849a7554075".to_string(),
+    );
+    let upstreams = vec![jd_client::proxy_config::Upstream {
+        authority_pubkey,
+        pool_address,
+        jd_address,
+        pool_signature,
+    }];
+    let config = jd_client::ProxyConfig::new(
+        downstream_address,
+        downstream_port,
+        max_supported_version,
+        min_supported_version,
+        min_extranonce2_size,
+        withhold,
+        authority_pubkey,
+        authority_secret_key,
+        cert_validity_sec,
+        tp_address,
+        Some(tp_authority_public_key),
+        retry,
+        upstreams,
+        std::time::Duration::from_secs(300),
+        vec![coinbase_outputs],
+    );
+    let task_collector = Arc::new(jd_client::Mutex::new(Vec::new()));
+    let client = jd_client::JobDeclaratorClient::new(config.clone(), task_collector);
+    (client, config)
+}
+
+pub async fn start_sv2_pool() {
+    use pool_sv2::mining_pool::{CoinbaseOutput, Configuration};
+    let authority_public_key = Secp256k1PublicKey::try_from(
+        "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72".to_string(),
+    )
+    .expect("failed");
+    let authority_secret_key = Secp256k1SecretKey::try_from(
+        "mkDLTBBRxdBv998612qipDYoTK3YUrqLe8uWw7gu3iXbSrn2n".to_string(),
+    )
+    .expect("failed");
+    let cert_validity_sec = 3600;
+    let listen_address = "0.0.0.0:34254".to_string();
+    let coinbase_outputs = vec![CoinbaseOutput::new(
+        "P2WPKH".to_string(),
+        "036adc3bdf21e6f9a0f0fb0066bf517e5b7909ed1563d6958a10993849a7554075".to_string(),
+    )];
+    let pool_signature = "Stratum v2 SRI Pool".to_string();
+    let tp_address = "75.119.150.111:8442".to_string();
+    let tp_authority_public_key = Secp256k1PublicKey::try_from(
+        "9azQdassggC7L3YMVcZyRJmK7qrFDj5MZNHb4LkaUrJRUhct92W".to_string(),
+    )
+    .expect("failed");
+    let config = Configuration::new(
+        listen_address,
+        tp_address,
+        Some(tp_authority_public_key),
+        authority_public_key,
+        authority_secret_key,
+        cert_validity_sec,
+        coinbase_outputs,
+        pool_signature,
+    );
+    tokio::spawn(async move {
+        pool_sv2::PoolSv2::start(config).await;
+    });
+}
+
+async fn connect_to_jd_client(address: String) -> Result<TcpStream, ()> {
+    loop {
+        match TcpStream::connect(address.clone()).await {
+            Ok(stream) => return Ok(stream),
+            Err(_e) => {
+                continue;
+            }
+        }
+    }
+}
+
+pub async fn start_sv2_translator(jd_client_address: String) -> translator_sv2::TranslatorSv2 {
+    let stream = connect_to_jd_client(jd_client_address).await;
+    let translator_v2 = translator_sv2::TranslatorSv2::new(stream.unwrap());
+    let upstream_address = "127.0.0.1".to_string();
+    let upstream_port = 34265;
+    let upstream_authority_pubkey = Secp256k1PublicKey::try_from(
+        "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72".to_string(),
+    )
+    .expect("failed");
+    let downstream_address = "0.0.0.0".to_string();
+    let downstream_port = 34255;
+    let max_supported_version = 2;
+    let min_supported_version = 2;
+    let min_extranonce2_size = 8;
+    let min_individual_miner_hashrate = 10_000_000_000_000.0;
+    let shares_per_minute = 6.0;
+    let channel_diff_update_interval = 60;
+    let channel_nominal_hashrate = 10_000_000_000_000.0;
+    let downstream_difficulty_config =
+        translator_sv2::proxy_config::DownstreamDifficultyConfig::new(
+            min_individual_miner_hashrate,
+            shares_per_minute,
+            0,
+            0,
+        );
+    let upstream_difficulty_config = translator_sv2::proxy_config::UpstreamDifficultyConfig::new(
+        channel_diff_update_interval,
+        channel_nominal_hashrate,
+        0,
+        false,
+    );
+    let config = translator_sv2::proxy_config::ProxyConfig::new(
+        upstream_address,
+        upstream_port,
+        upstream_authority_pubkey,
+        downstream_address,
+        downstream_port,
+        max_supported_version,
+        min_supported_version,
+        min_extranonce2_size,
+        downstream_difficulty_config,
+        upstream_difficulty_config,
+    );
+
+    let (tx_status, _rx_status) = async_channel::unbounded();
+    translator_v2.start(config, tx_status).await;
+    translator_v2
+}

roles/tests-integration/tests/integration_e2e.rs

@@ -0,0 +1,36 @@
+use std::{net::SocketAddr, str::FromStr};
+
+mod common;
+
+#[tokio::test]
+async fn test_jd_client_translator_sv2_pool_sv2_integration() {
+    common::start_sv2_pool().await;
+    common::start_job_declarator_server().await;
+    let (jd_client, jd_client_config) = common::start_job_declarator_client().await;
+    let upstream = jd_client_config.upstreams.clone();
+    let upstream = upstream.get(0).cloned().unwrap();
+    let address = format!(
+        "{}:{}",
+        jd_client_config.downstream_address, jd_client_config.downstream_port
+    );
+    let pool_addr =
+        SocketAddr::from_str(upstream.pool_address.as_str()).expect("Invalid pool address");
+    let pool_socket = {
+        loop {
+            match tokio::net::TcpStream::connect(pool_addr).await {
+                Ok(s) => break s,
+                Err(_e) => {
+                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                }
+            }
+        }
+    };
+    jd_client
+        .clone()
+        .initialize_jd(upstream.clone(), pool_socket)
+        .await;
+    let translator_sv2 = common::start_sv2_translator(address.clone()).await;
+    let upstream_address = translator_sv2.upstream_address().unwrap();
+    dbg!("Here");
+    assert_eq!(jd_client.downstream_address(), upstream_address);
+}