Move watch channel helpers to separate module inside test harness

Introduce `harness::flow` module, grouping helpers that await for the next value
on the watch channel.

daemon/tests/happy_path.rs

@@ -1,16 +1,13 @@
+use crate::harness::flow::{is_next_none, next_cfd, next_order, next_some};
 use crate::harness::mocks::{
     mock_oracle_annoucement, mock_party_params, mock_wallet_sign_and_broadcast,
 };
 use crate::harness::start_both;
-use anyhow::Context;
 use daemon::maker_cfd;
-use daemon::model::cfd::{Cfd, CfdState, Order, Origin};
+use daemon::model::cfd::{CfdState, Order, Origin};
 use daemon::model::{Price, Usd};
-use daemon::tokio_ext::FutureExt;
 use maia::secp256k1_zkp::schnorrsig;
 use rust_decimal_macros::dec;
-use std::time::Duration;
-use tokio::sync::watch;
 use tracing::subscriber::DefaultGuard;
 use tracing_subscriber::filter::LevelFilter;
 use tracing_subscriber::util::SubscriberInitExt;
@@ -130,69 +127,6 @@ fn dummy_new_order() -> maker_cfd::NewOrder {
     }
 }
 
-/// Returns the first `Cfd` from both channels
-///
-/// Ensures that there is only one `Cfd` present in both channels.
-async fn next_cfd(
-    rx_a: &mut watch::Receiver<Vec<Cfd>>,
-    rx_b: &mut watch::Receiver<Vec<Cfd>>,
-) -> (Cfd, Cfd) {
-    let (a, b) = tokio::join!(next(rx_a), next(rx_b));
-
-    assert_eq!(a.len(), 1);
-    assert_eq!(b.len(), 1);
-
-    (a.first().unwrap().clone(), b.first().unwrap().clone())
-}
-
-async fn next_order(
-    rx_a: &mut watch::Receiver<Option<Order>>,
-    rx_b: &mut watch::Receiver<Option<Order>>,
-) -> (Order, Order) {
-    let (a, b) = tokio::join!(next_some(rx_a), next_some(rx_b));
-
-    (a, b)
-}
-
-/// Returns the value if the next Option received on the stream is Some
-///
-/// Panics if None is received on the stream.
-async fn next_some<T>(rx: &mut watch::Receiver<Option<T>>) -> T
-where
-    T: Clone,
-{
-    if let Some(value) = next(rx).await {
-        value
-    } else {
-        panic!("Received None when Some was expected")
-    }
-}
-
-/// Returns true if the next Option received on the stream is None
-///
-/// Returns false if Some is received.
-async fn is_next_none<T>(rx: &mut watch::Receiver<Option<T>>) -> bool
-where
-    T: Clone,
-{
-    next(rx).await.is_none()
-}
-
-/// Returns watch channel value upon change
-async fn next<T>(rx: &mut watch::Receiver<T>) -> T
-where
-    T: Clone,
-{
-    // TODO: Make timeout configurable, only contract setup can take up to 2 min on CI
-    rx.changed()
-        .timeout(Duration::from_secs(120))
-        .await
-        .context("Waiting for next element in channel is taking too long, aborting")
-        .unwrap()
-        .unwrap();
-    rx.borrow().clone()
-}
-
 fn init_tracing() -> DefaultGuard {
     let filter = EnvFilter::from_default_env()
         // apply warning level globally

daemon/tests/harness/flow.rs

@@ -0,0 +1,69 @@
+use std::time::Duration;
+
+use anyhow::Context;
+use daemon::model::cfd::{Cfd, Order};
+use daemon::tokio_ext::FutureExt;
+use tokio::sync::watch;
+
+/// Returns the first `Cfd` from both channels
+///
+/// Ensures that there is only one `Cfd` present in both channels.
+pub async fn next_cfd(
+    rx_a: &mut watch::Receiver<Vec<Cfd>>,
+    rx_b: &mut watch::Receiver<Vec<Cfd>>,
+) -> (Cfd, Cfd) {
+    let (a, b) = tokio::join!(next(rx_a), next(rx_b));
+
+    assert_eq!(a.len(), 1);
+    assert_eq!(b.len(), 1);
+
+    (a.first().unwrap().clone(), b.first().unwrap().clone())
+}
+
+pub async fn next_order(
+    rx_a: &mut watch::Receiver<Option<Order>>,
+    rx_b: &mut watch::Receiver<Option<Order>>,
+) -> (Order, Order) {
+    let (a, b) = tokio::join!(next_some(rx_a), next_some(rx_b));
+
+    (a, b)
+}
+
+/// Returns the value if the next Option received on the stream is Some
+///
+/// Panics if None is received on the stream.
+pub async fn next_some<T>(rx: &mut watch::Receiver<Option<T>>) -> T
+where
+    T: Clone,
+{
+    if let Some(value) = next(rx).await {
+        value
+    } else {
+        panic!("Received None when Some was expected")
+    }
+}
+
+/// Returns true if the next Option received on the stream is None
+///
+/// Returns false if Some is received.
+pub async fn is_next_none<T>(rx: &mut watch::Receiver<Option<T>>) -> bool
+where
+    T: Clone,
+{
+    next(rx).await.is_none()
+}
+
+/// Returns watch channel value upon change
+pub async fn next<T>(rx: &mut watch::Receiver<T>) -> T
+where
+    T: Clone,
+{
+    // TODO: Make timeout configurable, only contract setup can take up to 2 min on CI
+    rx.changed()
+        .timeout(Duration::from_secs(120))
+        .await
+        .context("Waiting for next element in channel is taking too long, aborting")
+        .unwrap()
+        .unwrap();
+    rx.borrow().clone()
+}

daemon/tests/harness/mod.rs

@@ -16,6 +16,7 @@ use xtra::spawn::TokioGlobalSpawnExt;
 use xtra::Actor;
 
 pub mod bdk;
+pub mod flow;
 pub mod maia;
 pub mod mocks;