`unwrap` only in the tests

This makes the stacktrace more useful because it shows us where exactly
it is failing.

daemon/tests/happy_path.rs

@@ -11,14 +11,14 @@ async fn taker_receives_order_from_maker_on_publication() {
     let _guard = init_tracing();
     let (mut maker, mut taker) = start_both().await;
 
-    assert!(is_next_none(taker.order_feed()).await);
+    assert!(is_next_none(taker.order_feed()).await.unwrap());
 
     maker.publish_order(dummy_new_order()).await;
 
     let (published, received) =
         tokio::join!(next_some(maker.order_feed()), next_some(taker.order_feed()));
 
-    assert_is_same_order(&published, &received);
+    assert_is_same_order(&published.unwrap(), &received.unwrap());
 }
 
 #[tokio::test]
@@ -27,15 +27,17 @@ async fn taker_takes_order_and_maker_rejects() {
     let (mut maker, mut taker) = start_both().await;
 
     // TODO: Why is this needed? For the cfd stream it is not needed
-    is_next_none(taker.order_feed()).await;
+    is_next_none(taker.order_feed()).await.unwrap();
 
     maker.publish_order(dummy_new_order()).await;
 
-    let (_, received) = next_order(maker.order_feed(), taker.order_feed()).await;
+    let (_, received) = next_order(maker.order_feed(), taker.order_feed())
+        .await
+        .unwrap();
 
     taker.take_order(received.clone(), Usd::new(dec!(10))).await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await.unwrap();
     assert_is_same_order(&taker_cfd.order, &received);
     assert_is_same_order(&maker_cfd.order, &received);
     assert!(matches!(
@@ -49,7 +51,7 @@ async fn taker_takes_order_and_maker_rejects() {
 
     maker.reject_take_request(received.clone()).await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await.unwrap();
     // TODO: More elaborate Cfd assertions
     assert_is_same_order(&taker_cfd.order, &received);
     assert_is_same_order(&maker_cfd.order, &received);
@@ -62,14 +64,16 @@ async fn taker_takes_order_and_maker_accepts_and_contract_setup() {
     let _guard = init_tracing();
     let (mut maker, mut taker) = start_both().await;
 
-    is_next_none(taker.order_feed()).await;
+    is_next_none(taker.order_feed()).await.unwrap();
 
     maker.publish_order(dummy_new_order()).await;
 
-    let (_, received) = next_order(maker.order_feed(), taker.order_feed()).await;
+    let (_, received) = next_order(maker.order_feed(), taker.order_feed())
+        .await
+        .unwrap();
 
     taker.take_order(received.clone(), Usd::new(dec!(5))).await;
-    let (_, _) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
+    let (_, _) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await.unwrap();
 
     maker.mocks.mock_oracle_annoucement().await;
     taker.mocks.mock_oracle_annoucement().await;
@@ -79,7 +83,7 @@ async fn taker_takes_order_and_maker_accepts_and_contract_setup() {
 
     maker.accept_take_request(received.clone()).await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await.unwrap();
     // TODO: More elaborate Cfd assertions
     assert_eq!(taker_cfd.order.id, received.id);
     assert_eq!(maker_cfd.order.id, received.id);
@@ -89,7 +93,7 @@ async fn taker_takes_order_and_maker_accepts_and_contract_setup() {
     maker.mocks.mock_wallet_sign_and_broadcast().await;
     taker.mocks.mock_wallet_sign_and_broadcast().await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await.unwrap();
     // TODO: More elaborate Cfd assertions
     assert_eq!(taker_cfd.order.id, received.id);
     assert_eq!(maker_cfd.order.id, received.id);

daemon/tests/harness/flow.rs

@@ -1,8 +1,7 @@
-use std::time::Duration;
+use anyhow::{Context, Result};
-
-use anyhow::Context;
 use daemon::model::cfd::{Cfd, Order};
 use daemon::tokio_ext::FutureExt;
+use std::time::Duration;
 use tokio::sync::watch;
 
 /// Returns the first `Cfd` from both channels
@@ -11,58 +10,54 @@ use tokio::sync::watch;
 pub async fn next_cfd(
     rx_a: &mut watch::Receiver<Vec<Cfd>>,
     rx_b: &mut watch::Receiver<Vec<Cfd>>,
-) -> (Cfd, Cfd) {
+) -> Result<(Cfd, Cfd)> {
     let (a, b) = tokio::join!(next(rx_a), next(rx_b));
+    let (a, b) = (a?, b?);
 
     assert_eq!(a.len(), 1);
     assert_eq!(b.len(), 1);
 
-    (a.first().unwrap().clone(), b.first().unwrap().clone())
+    Ok((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) {
+) -> Result<(Order, Order)> {
     let (a, b) = tokio::join!(next_some(rx_a), next_some(rx_b));
 
-    (a, b)
+    Ok((a?, b?))
 }
 
 /// Returns the value if the next Option received on the stream is Some
-///
+pub async fn next_some<T>(rx: &mut watch::Receiver<Option<T>>) -> Result<T>
-/// 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 {
+    next(rx)
-        value
+        .await?
-    } else {
+        .context("Received None when Some was expected")
-        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
+pub async fn is_next_none<T>(rx: &mut watch::Receiver<Option<T>>) -> Result<bool>
 where
     T: Clone,
 {
-    next(rx).await.is_none()
+    Ok(next(rx).await?.is_none())
 }
 
 /// Returns watch channel value upon change
-pub async fn next<T>(rx: &mut watch::Receiver<T>) -> T
+pub async fn next<T>(rx: &mut watch::Receiver<T>) -> Result<T>
 where
     T: Clone,
 {
     rx.changed()
         .timeout(Duration::from_secs(10))
         .await
-        .context("Waiting for next element in channel is taking too long, aborting")
+        .context("No change in channel within 10 seconds")??;
-        .unwrap()
+
-        .unwrap();
+    Ok(rx.borrow().clone())
-    rx.borrow().clone()
 }