Repository: farcaster-project/farcaster-node
Pull request #284 (2021-12-02 10:02:29)

Add parallel swap test
Adds a functional tests that runs four swaps in parallel (2 with Bob maker, 2 with Alice maker).

codecov-commenter: # [Codecov](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#284](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (173d656) into [main](https://codecov.io/gh/farcaster-project/farcaster-node/commit/eeaab30a5948d71a4533094a500eb0b1528032fb?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (eeaab30) will **increase** coverage by `0.0%`.
> The diff coverage is `0.0%`.

[![Impacted file tree graph](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284/graphs/tree.svg?width=650&height=150&src=pr&token=ZZ9RnFhgFC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@          Coverage Diff          @@
##            main    #284   +/-   ##
=====================================
  Coverage   15.3%   15.3%           
=====================================
  Files         27      27           
  Lines       6550    6548    -2     
=====================================
  Hits        1003    1003           
+ Misses      5547    5545    -2     
```


| [Impacted Files](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [src/farcasterd/runtime.rs](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-c3JjL2ZhcmNhc3RlcmQvcnVudGltZS5ycw==) | `0.0% <0.0%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None). Last update [eeaab30...173d656](https://codecov.io/gh/farcaster-project/farcaster-node/pull/284?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).


Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 034421eb..e0d7592e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -957,6 +957,7 @@ dependencies = [
  "electrum-client",
  "env_logger",
  "farcaster_core",
+ "futures",
  "hex",
  "internet2",
  "lazy_static",
@@ -970,6 +971,7 @@ dependencies = [
  "nix",
  "ntest",
  "paste",
+ "rand 0.8.4",
  "regex",
  "serde 1.0.127",
  "serde_json",
@@ -1049,9 +1051,9 @@ checksum = "1847abb9cb65d566acd5942e94aea9c8f547ad02c98e1649326fc0e8910b8b1e"
 
 [[package]]
 name = "futures"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1adc00f486adfc9ce99f77d717836f0c5aa84965eb0b4f051f4e83f7cab53f8b"
+checksum = "8cd0210d8c325c245ff06fd95a3b13689a1a276ac8cfa8e8720cb840bfb84b9e"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -1064,9 +1066,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "74ed2411805f6e4e3d9bc904c95d5d423b89b3b25dc0250aa74729de20629ff9"
+checksum = "7fc8cd39e3dbf865f7340dce6a2d401d24fd37c6fe6c4f0ee0de8bfca2252d27"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -1074,15 +1076,15 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "af51b1b4a7fdff033703db39de8802c673eb91855f2e0d47dcf3bf2c0ef01f99"
+checksum = "629316e42fe7c2a0b9a65b47d159ceaa5453ab14e8f0a3c5eedbb8cd55b4a445"
 
 [[package]]
 name = "futures-executor"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d0d535a57b87e1ae31437b892713aee90cd2d7b0ee48727cd11fc72ef54761c"
+checksum = "7b808bf53348a36cab739d7e04755909b9fcaaa69b7d7e588b37b6ec62704c97"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -1091,18 +1093,16 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b0e06c393068f3a6ef246c75cdca793d6a46347e75286933e5e75fd2fd11582"
+checksum = "e481354db6b5c353246ccf6a728b0c5511d752c08da7260546fc0933869daa11"
 
 [[package]]
 name = "futures-macro"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c54913bae956fb8df7f4dc6fc90362aa72e69148e3f39041fbe8742d21e0ac57"
+checksum = "a89f17b21645bc4ed773c69af9c9a0effd4a3f1a3876eadd453469f8854e7fdd"
 dependencies = [
- "autocfg 1.0.1",
- "proc-macro-hack",
  "proc-macro2",
  "quote",
  "syn",
@@ -1110,23 +1110,22 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c0f30aaa67363d119812743aa5f33c201a7a66329f97d1a887022971feea4b53"
+checksum = "996c6442437b62d21a32cd9906f9c41e7dc1e19a9579843fad948696769305af"
 
 [[package]]
 name = "futures-task"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bbe54a98670017f3be909561f6ad13e810d9a51f3f061b902062ca3da80799f2"
+checksum = "dabf1872aaab32c886832f2276d2f5399887e2bd613698a02359e4ea83f8de12"
 
 [[package]]
 name = "futures-util"
-version = "0.3.16"
+version = "0.3.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "67eb846bfd58e44a8481a00049e82c43e0ccb5d61f8dc071057cb19249dd4d78"
+checksum = "41d22213122356472061ac0f1ab2cee28d2bac8491410fd68c2af53d1cedb83e"
 dependencies = [
- "autocfg 1.0.1",
  "futures-channel",
  "futures-core",
  "futures-io",
@@ -1136,8 +1135,6 @@ dependencies = [
  "memchr",
  "pin-project-lite",
  "pin-utils",
- "proc-macro-hack",
- "proc-macro-nested",
  "slab",
 ]
 
@@ -2070,18 +2067,6 @@ dependencies = [
  "version_check",
 ]
 
-[[package]]
-name = "proc-macro-hack"
-version = "0.5.19"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dbf0c48bc1d91375ae5c3cd81e3722dff1abcf81a30960240640d223f59fe0e5"
-
-[[package]]
-name = "proc-macro-nested"
-version = "0.1.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc881b2c22681370c6a780e47af9840ef841837bc98118431d4e1868bd0c1086"
-
 [[package]]
 name = "proc-macro2"
 version = "1.0.28"
diff --git a/Cargo.toml b/Cargo.toml
index adf3de98..d5256f87 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -130,9 +130,11 @@ features = ['peer']
 
 [dev-dependencies]
 ntest = "0.7.3"
+rand = "0.8.4"
 
 # Test
 strip-ansi-escapes = "0.1.1"
+futures = "0.3.18"
 
 [build-dependencies.microservices]
 # path = '../ext/rust-internet2'


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 34d770b4..51b7c0f0 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -47,8 +47,8 @@ jobs:
           file: ${{ steps.coverage.outputs.report }}
           directory: ./coverage/reports/
 
-  functional-test:
-    name: Functional tests
+  functional-syncer-test:
+    name: Functional syncer tests
     strategy:
       matrix:
         test: [
@@ -82,3 +82,31 @@ jobs:
         with:
           command: test
           args: ${{ matrix.test }} --workspace --all-targets --all-features --no-fail-fast --verbose -- --ignored --test-threads=1
+
+  functional-swap-test:
+    name: Functional swap tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install dependencies
+        run: sudo apt-get install -y cmake libzmq3-dev docker-compose
+
+      - name: Compose docker services
+        run: cd tests && docker-compose up -d
+
+      - name: Install Rust stable
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+
+      - name: Correct permissions
+        run: sudo chown -R $USER tests
+
+      - name: Rust cache
+        uses: Swatinem/rust-cache@v1.3.0
+
+      - name: Cargo test
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: swap_test --workspace --all-targets --all-features --no-fail-fast --verbose -- --ignored --test-threads=1
\ No newline at end of file
diff --git a/tests/functional-swap.rs b/tests/functional-swap.rs
index e5400f68..3aeac67a 100644
--- a/tests/functional-swap.rs
+++ b/tests/functional-swap.rs
@@ -2,44 +2,57 @@
 extern crate log;
 
 use bitcoincore_rpc::{Auth, Client, RpcApi};
+use farcaster_node::rpc::request::NodeInfo;
+use futures::future::join_all;
+use std::collections::HashSet;
 use std::ffi::OsStr;
 use std::io;
 use std::process;
-use std::{thread, time};
+use std::sync::Arc;
+use std::time;
 use sysinfo::{ProcessExt, System, SystemExt};
+use tokio::sync::Mutex;
 
 use std::collections::HashMap;
 use std::str;
 use std::str::FromStr;
 
+use ntest::timeout;
+
 const ALLOWED_RETRIES: u32 = 120;
 
 #[tokio::test]
+#[timeout(600000)]
 #[ignore]
 async fn swap_test_bob_maker() {
-    let bitcoin_rpc = bitcoin_setup();
+    let execution_mutex = Arc::new(Mutex::new(0));
+    let bitcoin_rpc = Arc::new(bitcoin_setup());
     let (monero_regtest, monero_wallet) = monero_setup().await;
 
     let (farcasterd_maker, data_dir_maker, farcasterd_taker, data_dir_taker) =
         setup_farcaster_clients().await;
 
-    let (_, bitcoin_address, swap_ids) = make_and_take_offer(
+    let (xmr_dest_wallet_name, bitcoin_address, swap_id) = make_and_take_offer(
         data_dir_maker.clone(),
         data_dir_taker.clone(),
         "Bob".to_string(),
-        &bitcoin_rpc,
-        &monero_wallet,
+        Arc::clone(&bitcoin_rpc),
+        Arc::clone(&monero_wallet),
+        "1 BTC".to_string(),
+        "1 XMR".to_string(),
     )
     .await;
 
     run_swap(
-        swap_ids[0].clone(),
+        swap_id,
         data_dir_taker,
         data_dir_maker,
-        bitcoin_rpc,
+        Arc::clone(&bitcoin_rpc),
         bitcoin_address,
         monero_regtest,
-        monero_wallet,
+        Arc::clone(&monero_wallet),
+        xmr_dest_wallet_name,
+        execution_mutex,
     )
     .await;
 
@@ -47,48 +60,162 @@ async fn swap_test_bob_maker() {
 }
 
 #[tokio::test]
+#[timeout(600000)]
 #[ignore]
 async fn swap_test_alice_maker() {
-    let bitcoin_rpc = bitcoin_setup();
+    let execution_mutex = Arc::new(Mutex::new(0));
+    let bitcoin_rpc = Arc::new(bitcoin_setup());
     let (monero_regtest, monero_wallet) = monero_setup().await;
 
     let (farcasterd_maker, data_dir_maker, farcasterd_taker, data_dir_taker) =
         setup_farcaster_clients().await;
 
-    let (_, bitcoin_address, swap_ids) = make_and_take_offer(
+    let (xmr_dest_wallet_name, bitcoin_address, swap_id) = make_and_take_offer(
         data_dir_maker.clone(),
         data_dir_taker.clone(),
         "Alice".to_string(),
-        &bitcoin_rpc,
-        &monero_wallet,
+        Arc::clone(&bitcoin_rpc),
+        Arc::clone(&monero_wallet),
+        "1 BTC".to_string(),
+        "1 XMR".to_string(),
     )
     .await;
 
     run_swap(
-        swap_ids[0].clone(),
+        swap_id,
         data_dir_maker,
         data_dir_taker,
-        bitcoin_rpc,
+        Arc::clone(&bitcoin_rpc),
         bitcoin_address,
         monero_regtest,
-        monero_wallet,
+        Arc::clone(&monero_wallet),
+        xmr_dest_wallet_name,
+        execution_mutex,
     )
     .await;
 
     cleanup_processes(farcasterd_maker, farcasterd_taker);
 }
 
+#[tokio::test]
+#[timeout(600000)]
+#[ignore]
+async fn swap_test_parallel() {
+    let execution_mutex = Arc::new(Mutex::new(0));
+    let bitcoin_rpc = Arc::new(bitcoin_setup());
+    let (monero_regtest, monero_wallet) = monero_setup().await;
+
+    let (farcasterd_maker, data_dir_maker, farcasterd_taker, data_dir_taker) =
+        setup_farcaster_clients().await;
+
+    let (xmr_dest_wallet_name, bitcoin_address, swap_id) = make_and_take_offer(
+        data_dir_maker.clone(),
+        data_dir_taker.clone(),
+        "Alice".to_string(),
+        Arc::clone(&bitcoin_rpc),
+        Arc::clone(&monero_wallet),
+        "1 BTC".to_string(),
+        "1 XMR".to_string(),
+    )
+    .await;
+
+    let alice_future = run_swap(
+        swap_id,
+        data_dir_maker.clone(),
+        data_dir_taker.clone(),
+        Arc::clone(&bitcoin_rpc),
+        bitcoin_address.clone(),
+        monero_regtest.clone(),
+        Arc::clone(&monero_wallet),
+        xmr_dest_wallet_name,
+        Arc::clone(&execution_mutex),
+    );
+
+    let (xmr_dest_wallet_name, bitcoin_address, swap_id) = make_and_take_offer(
+        data_dir_maker.clone(),
+        data_dir_taker.clone(),
+        "Alice".to_string(),
+        Arc::clone(&bitcoin_rpc),
+        Arc::clone(&monero_wallet),
+        "1 BTC".to_string(),
+        "0.999999999999 XMR".to_string(),
+    )
+    .await;
+
+    let alice_future_1 = run_swap(
+        swap_id,
+        data_dir_maker.clone(),
+        data_dir_taker.clone(),
+        Arc::clone(&bitcoin_rpc),
+        bitcoin_address.clone(),
+        monero_regtest.clone(),
+        Arc::clone(&monero_wallet),
+        xmr_dest_wallet_name,
+        Arc::clone(&execution_mutex),
+    );
+
+    let (xmr_dest_wallet_name, bitcoin_address, swap_id) = make_and_take_offer(
+        data_dir_maker.clone(),
+        data_dir_taker.clone(),
+        "Bob".to_string(),
+        Arc::clone(&bitcoin_rpc),
+        Arc::clone(&monero_wallet),
+        "1 BTC".to_string(),
+        "1 XMR".to_string(),
+    )
+    .await;
+
+    let bob_future = run_swap(
+        swap_id,
+        data_dir_taker.clone(),
+        data_dir_maker.clone(),
+        Arc::clone(&bitcoin_rpc),
+        bitcoin_address,
+        monero_regtest.clone(),
+        Arc::clone(&monero_wallet.clone()),
+        xmr_dest_wallet_name,
+        Arc::clone(&execution_mutex),
+    );
+
+    let (xmr_dest_wallet_name, bitcoin_address, swap_id) = make_and_take_offer(
+        data_dir_maker.clone(),
+        data_dir_taker.clone(),
+        "Bob".to_string(),
+        Arc::clone(&bitcoin_rpc),
+        Arc::clone(&monero_wallet),
+        "1 BTC".to_string(),
+        "0.999999999999 XMR".to_string(),
+    )
+    .await;
+
+    let bob_future_1 = run_swap(
+        swap_id,
+        data_dir_taker.clone(),
+        data_dir_maker.clone(),
+        Arc::clone(&bitcoin_rpc),
+        bitcoin_address,
+        monero_regtest.clone(),
+        Arc::clone(&monero_wallet.clone()),
+        xmr_dest_wallet_name,
+        Arc::clone(&execution_mutex),
+    );
+
+    join_all(vec![alice_future, alice_future_1, bob_future, bob_future_1]).await;
+
+    cleanup_processes(farcasterd_maker, farcasterd_taker);
+}
+
 async fn setup_farcaster_clients() -> (process::Child, Vec<String>, process::Child, Vec<String>) {
     // data directories
     let data_dir_maker = vec!["-d".to_string(), "tests/.farcaster_node_0".to_string()];
     let data_dir_taker = vec!["-d".to_string(), "tests/.farcaster_node_1".to_string()];
 
-    let server_args_maker = vec!["-vvv", "--config", "tests/.farcasterd_1.toml"]
+    let server_args_maker = vec!["-vv", "--config", "tests/.farcasterd_1.toml"]
         .iter()
         .map(|i| i.to_string())
         .collect();
 
-    let server_args_taker = vec!["-vvv", "--config", "tests/.farcasterd_2.toml"]
+    let server_args_taker = vec!["-vv", "--config", "tests/.farcasterd_2.toml"]
         .iter()
         .map(|i| i.to_string())
         .collect();
@@ -110,30 +237,49 @@ async fn make_and_take_offer(
     data_dir_maker: Vec<String>,
     data_dir_taker: Vec<String>,
     role: String,
-    bitcoin_rpc: &bitcoincore_rpc::Client,
-    monero_wallet: &monero_rpc::WalletClient,
-) -> (monero::Address, bitcoin::Address, Vec<String>) {
+    bitcoin_rpc: Arc<bitcoincore_rpc::Client>,
+    monero_wallet: Arc<Mutex<monero_rpc::WalletClient>>,
+    btc_amount: String,
+    xmr_amount: String,
+) -> (String, bitcoin::Address, String) {
     let maker_info_args = info_args(data_dir_maker.clone());
+    let taker_info_args = info_args(data_dir_maker.clone());
 
     // test connection to farcasterd and check that swap-cli is in the correct place
     run("../swap-cli", maker_info_args.clone()).unwrap();
 
-    let xmr_address = monero_wallet.get_address(0, None).await.unwrap().address;
+    let (xmr_address, xmr_address_wallet_name) =
+        monero_new_dest_address(Arc::clone(&monero_wallet)).await;
     let btc_address = bitcoin_rpc.get_new_address(None, None).unwrap();
-
     let btc_addr = btc_address.to_string();
     let xmr_addr = xmr_address.to_string();
 
+    let (stdout, _stderr) = run("../swap-cli", taker_info_args.clone()).unwrap();
+    let previous_swap_ids: HashSet<String> = stdout
+        .iter()
+        .filter_map(|element| {
+            if let Some(pos) = element.find("\"0x") {
+                let len = element.to_string().len() - 1;
+                let swap_id = element[pos + 1..len].to_string();
+                Some(swap_id)
+            } else {
+                None
+            }
+        })
+        .collect();
+
     let cli_make_args = make_offer_args(
         data_dir_maker.clone(),
         role,
         btc_addr.clone(),
+        btc_amount,
         xmr_addr.clone(),
+        xmr_amount,
     );
     let (_stdout, _stderr) = run("../swap-cli", cli_make_args).unwrap();
 
     // get offer strings
-    let offers = retry_until_offer(maker_info_args.clone());
+    let offers = retry_until_offer(maker_info_args.clone()).await;
 
     let cli_take_args = take_offer_args(
         data_dir_taker.clone(),
@@ -143,19 +289,22 @@ async fn make_and_take_offer(
     );
     run("../swap-cli", cli_take_args).unwrap();
 
-    let swap_ids = retry_until_swap_ids(maker_info_args.clone());
+    let swap_id = retry_until_swap_id(taker_info_args.clone(), previous_swap_ids).await;
 
-    (xmr_address, btc_address, swap_ids)
+    (xmr_address_wallet_name, btc_address, swap_id)
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn run_swap(
     swap_id: String,
     data_dir_alice: Vec<String>,
     data_dir_bob: Vec<String>,
-    bitcoin_rpc: bitcoincore_rpc::Client,
+    bitcoin_rpc: Arc<bitcoincore_rpc::Client>,
     funding_btc_address: bitcoin::Address,
     monero_regtest: monero_rpc::RegtestDaemonClient,
-    monero_wallet: monero_rpc::WalletClient,
+    monero_wallet: Arc<Mutex<monero_rpc::WalletClient>>,
+    monero_dest_wallet_name: String,
+    execution_mutex: Arc<Mutex<u8>>,
 ) {
     // let btc_address = bitcoin_rpc.get_new_address(None, None).unwrap();
     let cli_alice_progress_args: Vec<String> = progress_args(data_dir_alice, swap_id.clone());
@@ -166,9 +315,10 @@ async fn run_swap(
         .unwrap();
 
     // run until bob has the btc funding address
-    let address = retry_until_bitcoin_funding_address(cli_bob_progress_args.clone());
+    let address = retry_until_bitcoin_funding_address(cli_bob_progress_args.clone()).await;
 
     // fund the bitcoin address
+    let lock = execution_mutex.lock().await;
     let amount = bitcoin::Amount::ONE_SAT * 100000150;
     bitcoin_rpc
         .send_to_address(&address, amount, None, None, None, None, None, None)
@@ -179,8 +329,8 @@ async fn run_swap(
         .unwrap();
 
     // run until the alice has the monero funding address
-    let monero_address = retry_until_monero_funding_address(cli_alice_progress_args.clone());
-    send_monero(&monero_wallet, monero_address, 1000000000000).await;
+    let monero_address = retry_until_monero_funding_address(cli_alice_progress_args.clone()).await;
+    send_monero(Arc::clone(&monero_wallet), monero_address, 1000000000000).await;
 
     // generate some bitcoin blocks for confirmations
     bitcoin_rpc
@@ -191,7 +341,8 @@ async fn run_swap(
     retry_until_finish_state_transition(
         cli_alice_progress_args.clone(),
         "AliceState(Finish)".to_string(),
-    );
+    )
+    .await;
 
     // generate some blocks on bitcoin's side
     bitcoin_rpc
@@ -204,11 +355,17 @@ async fn run_swap(
     let balance = bitcoin_rpc
         .get_received_by_address(&funding_btc_address, None)
         .unwrap();
-    println!("balance: {:?}", balance.as_sat());
     assert!(balance.as_sat() > 90000000);
+    drop(lock);
 
     // cache the monero balance before sweeping
-    let before_balance = monero_wallet.get_balance(0, None).await.unwrap();
+    let monero_wallet_lock = monero_wallet.lock().await;
+    monero_wallet_lock
+        .open_wallet(monero_dest_wallet_name.clone(), None)
+        .await
+        .unwrap();
+    let before_balance = monero_wallet_lock.get_balance(0, None).await.unwrap();
+    drop(monero_wallet_lock);
 
     // generate some blocks on monero's side
     monero_regtest
@@ -220,15 +377,22 @@ async fn run_swap(
     retry_until_finish_state_transition(
         cli_bob_progress_args.clone(),
         "BobState(Finish)".to_string(),
-    );
+    )
+    .await;
 
     monero_regtest
         .generate_blocks(1, reusable_xmr_address())
         .await
         .unwrap();
 
-    monero_wallet.refresh(Some(1)).await.unwrap();
-    let after_balance = monero_wallet.get_balance(0, None).await.unwrap();
+    let monero_wallet_lock = monero_wallet.lock().await;
+    monero_wallet_lock
+        .open_wallet(monero_dest_wallet_name, None)
+        .await
+        .unwrap();
+    monero_wallet_lock.refresh(Some(1)).await.unwrap();
+    let after_balance = monero_wallet_lock.get_balance(0, None).await.unwrap();
+    drop(monero_wallet_lock);
     let delta_balance = after_balance.balance - before_balance.balance;
     assert!(delta_balance > 999660000000);
 }
@@ -283,7 +447,9 @@ fn make_offer_args(
     data_dir: Vec<String>,
     role: String,
     btc_addr: String,
+    btc_amount: String,
     xmr_addr: String,
+    xmr_amount: String,
 ) -> Vec<String> {
     data_dir
         .into_iter()
@@ -294,8 +460,8 @@ fn make_offer_args(
             "Local".to_string(),
             "ECDSA".to_string(),
             "Monero".to_string(),
-            "1 BTC".to_string(),
-            "1 XMR".to_string(),
+            btc_amount,
+            xmr_amount,
             role,
             "10".to_string(),
             "30".to_string(),
@@ -332,55 +498,49 @@ fn progress_args(data_dir: Vec<String>, swap_id: String) -> Vec<String> {
         .collect()
 }
 
-fn retry_until_offer(args: Vec<String>) -> Vec<String> {
+async fn retry_until_offer(args: Vec<String>) -> Vec<String> {
     for _ in 0..ALLOWED_RETRIES {
-        let (stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
-        let offers: Vec<String> = stdout
-            .iter()
-            .filter_map(|element| {
-                if let Some(pos) = element.find("Offer") {
-                    let len = element.to_string().len() - 1;
-                    let offer = element[pos..len].to_string();
-                    Some(offer)
-                } else {
-                    None
-                }
-            })
-            .collect();
+        let (mut stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
+        let info_str = stdout
+            .iter_mut()
+            .map(|line| format!("{}{}", line, "\n"))
+            .collect::<String>();
+        let info = serde_yaml::from_str::<NodeInfo>(&info_str).unwrap();
+        let offers: Vec<String> = info.offers.iter().map(|offer| offer.to_string()).collect();
         if !offers.is_empty() {
             return offers;
         }
-        thread::sleep(time::Duration::from_secs(1));
+        tokio::time::sleep(time::Duration::from_secs(1)).await;
     }
     panic!("timeout before any offer could be retrieved");
 }
 
-fn retry_until_swap_ids(args: Vec<String>) -> Vec<String> {
+async fn retry_until_swap_id(args: Vec<String>, previous_swap_ids: HashSet<String>) -> String {
     for _ in 0..ALLOWED_RETRIES {
-        let (stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
-        let swap_ids: Vec<String> = stdout
+        // let (stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
+        let (mut stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
+        let info_str = stdout
+            .iter_mut()
+            .map(|line| format!("{}{}", line, "\n"))
+            .collect::<String>();
+        let info = serde_yaml::from_str::<NodeInfo>(&info_str).unwrap();
+        let new_swap_ids: HashSet<String> = info
+            .swaps
             .iter()
-            .filter_map(|element| {
-                if let Some(pos) = element.find("\"0x") {
-                    let len = element.to_string().len() - 1;
-                    let swap_id = element[pos + 1..len].to_string();
-                    Some(swap_id)
-                } else {
-                    None
-                }
-            })
+            .map(|swap_id| format!("{:#x}", swap_id))
             .collect();
-        if !swap_ids.is_empty() {
-            return swap_ids;
+        if let Some(swap_id) = new_swap_ids.difference(&previous_swap_ids).next() {
+            return swap_id.to_string();
         }
-        thread::sleep(time::Duration::from_secs(1));
+        tokio::time::sleep(time::Duration::from_secs(1)).await;
     }
     panic!("timeout before any swapid could be retrieved");
 }
 
-fn retry_until_bitcoin_funding_address(args: Vec<String>) -> bitcoin::Address {
+async fn retry_until_bitcoin_funding_address(args: Vec<String>) -> bitcoin::Address {
     for _ in 0..ALLOWED_RETRIES {
         let (stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
+
         // get the btc funding address
         let bitcoin_address: Vec<String> = stdout
             .iter()
@@ -398,12 +558,12 @@ fn retry_until_bitcoin_funding_address(args: Vec<String>) -> bitcoin::Address {
         if !bitcoin_address.is_empty() {
             return bitcoin::Address::from_str(&bitcoin_address[0]).unwrap();
         }
-        thread::sleep(time::Duration::from_secs(1));
+        tokio::time::sleep(time::Duration::from_secs(1)).await;
     }
     panic!("timeout before any bitcoin funding address could be retrieved");
 }
 
-fn retry_until_monero_funding_address(args: Vec<String>) -> monero::Address {
+async fn retry_until_monero_funding_address(args: Vec<String>) -> monero::Address {
     for _ in 0..ALLOWED_RETRIES {
         let (stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
 
@@ -423,12 +583,15 @@ fn retry_until_monero_funding_address(args: Vec<String>) -> monero::Address {
         if !monero_addresses.is_empty() {
             return monero::Address::from_str(&monero_addresses[0]).unwrap();
         }
-        thread::sleep(time::Duration::from_secs(1));
+        tokio::time::sleep(time::Duration::from_secs(1)).await;
     }
     panic!("timeout before any monero funding address could be retrieved");
 }
 
-fn retry_until_finish_state_transition(args: Vec<String>, finish_state: String) -> Vec<String> {
+async fn retry_until_finish_state_transition(
+    args: Vec<String>,
+    finish_state: String,
+) -> Vec<String> {
     for _ in 0..ALLOWED_RETRIES {
         let (stdout, _stderr) = run("../swap-cli", args.clone()).unwrap();
 
@@ -447,7 +610,7 @@ fn retry_until_finish_state_transition(args: Vec<String>, finish_state: String)
             return alice_finish;
         }
 
-        thread::sleep(time::Duration::from_secs(1));
+        tokio::time::sleep(time::Duration::from_secs(1)).await;
     }
     panic!(
         "timeout before finish state {:?} could be retrieved",
@@ -472,17 +635,23 @@ fn bitcoin_setup() -> bitcoincore_rpc::Client {
     bitcoin_rpc
 }
 
-async fn monero_setup() -> (monero_rpc::RegtestDaemonClient, monero_rpc::WalletClient) {
+async fn monero_setup() -> (
+    monero_rpc::RegtestDaemonClient,
+    Arc<Mutex<monero_rpc::WalletClient>>,
+) {
     let daemon_client = monero_rpc::RpcClient::new("http://localhost:18081".to_string());
     let daemon = daemon_client.daemon();
     let regtest = daemon.regtest();
     let wallet_client = monero_rpc::RpcClient::new("http://localhost:18083".to_string());
     let wallet = wallet_client.wallet();
-    if let Err(_e) = wallet.open_wallet("test".to_string(), None).await {
-        wallet
+
+    if wallet.open_wallet("test".to_string(), None).await.is_err() {
+        // TODO: investigate this error in monero-rpc-rs
+        if wallet
             .create_wallet("test".to_string(), None, "English".to_string())
             .await
-            .unwrap();
+            .is_err()
+        {};
         wallet.open_wallet("test".to_string(), None).await.unwrap();
     }
 
@@ -493,7 +662,36 @@ async fn monero_setup() -> (monero_rpc::RegtestDaemonClient, monero_rpc::WalletC
         .await
         .unwrap();
 
-    (regtest, wallet)
+    (regtest, Arc::new(Mutex::new(wallet)))
+}
+
+async fn monero_new_dest_address(
+    wallet: Arc<Mutex<monero_rpc::WalletClient>>,
+) -> (monero::Address, String) {
+    use rand::distributions::Alphanumeric;
+    use rand::{thread_rng, Rng};
+
+    let wallet_name: String = thread_rng()
+        .sample_iter(&Alphanumeric)
+        .take(15)
+        .map(char::from)
+        .collect();
+
+    let wallet_lock = wallet.lock().await;
+
+    // TODO: investigate this error in monero-rpc-rs
+    if wallet_lock
+        .create_wallet(wallet_name.clone(), None, "English".to_string())
+        .await
+        .is_err()
+    {};
+    wallet_lock
+        .open_wallet(wallet_name.clone(), None)
+        .await
+        .unwrap();
+    let address = wallet_lock.get_address(0, None).await.unwrap();
+
+    (address.address, wallet_name)
 }
 
 pub fn run(
@@ -572,7 +770,7 @@ pub fn launch(
 }
 
 async fn send_monero(
-    wallet: &monero_rpc::WalletClient,
+    wallet: Arc<Mutex<monero_rpc::WalletClient>>,
     address: monero::Address,
     amount: u64,
 ) -> Vec<u8> {
@@ -587,7 +785,12 @@ async fn send_monero(
         payment_id: None,
         do_not_relay: None,
     };
-    let transaction = wallet
+    let wallet_lock = wallet.lock().await;
+    wallet_lock
+        .open_wallet("test".to_string(), None)
+        .await
+        .unwrap();
+    let transaction = wallet_lock
         .transfer(
             destination.clone(),
             monero_rpc::TransferPriority::Default,
@@ -595,5 +798,6 @@ async fn send_monero(
         )
         .await
         .unwrap();
+    drop(wallet_lock);
     hex::decode(transaction.tx_hash.to_string()).unwrap()
 }