Fix payjoin-cli v1 coverage

contrib/coverage.sh

@@ -4,4 +4,5 @@ set -e
 # https://github.com/taiki-e/cargo-llvm-cov?tab=readme-ov-file#merge-coverages-generated-under-different-test-conditions
 cargo llvm-cov clean --workspace # remove artifacts that may affect the coverage results
 cargo llvm-cov --no-report --all-features
+cargo llvm-cov --no-report --package payjoin-cli --no-default-features --features=v1,_danger-local-https # Explicitly run payjoin-cli v1 e2e tests
 cargo llvm-cov report --lcov --output-path lcov.info # generate report without tests

payjoin-cli/src/app/mod.rs

@@ -10,6 +10,8 @@ use payjoin::bitcoin::psbt::Psbt;
 use payjoin::bitcoin::FeeRate;
 use payjoin::receive::InputPair;
 use payjoin::{bitcoin, PjUri};
+use tokio::signal;
+use tokio::sync::watch;
 
 pub mod config;
 use crate::app::config::AppConfig;
@@ -136,3 +138,10 @@ pub fn input_pair_from_list_unspent(
     };
     InputPair::new(txin, psbtin).expect("Input pair should be valid")
 }
+
+async fn handle_interrupt(tx: watch::Sender<()>) {
+    if let Err(e) = signal::ctrl_c().await {
+        eprintln!("Error setting up Ctrl-C handler: {}", e);
+    }
+    let _ = tx.send(());
+}

payjoin-cli/src/app/v1.rs

@@ -20,10 +20,11 @@ use payjoin::receive::Error;
 use payjoin::send::v1::SenderBuilder;
 use payjoin::{Uri, UriExt};
 use tokio::net::TcpListener;
+use tokio::sync::watch;
 
 use super::config::AppConfig;
 use super::App as AppTrait;
-use crate::app::{http_agent, input_pair_from_list_unspent};
+use crate::app::{handle_interrupt, http_agent, input_pair_from_list_unspent};
 use crate::db::Database;
 #[cfg(feature = "_danger-local-https")]
 pub const LOCAL_CERT_FILE: &str = "localhost.der";
@@ -39,13 +40,16 @@ impl payjoin::receive::v1::Headers for Headers<'_> {
 pub(crate) struct App {
     config: AppConfig,
     db: Arc<Database>,
+    interrupt: watch::Receiver<()>,
 }
 
 #[async_trait::async_trait]
 impl AppTrait for App {
     fn new(config: AppConfig) -> Result<Self> {
         let db = Arc::new(Database::create(&config.db_path)?);
-        let app = Self { config, db };
+        let (interrupt_tx, interrupt_rx) = watch::channel(());
+        tokio::spawn(handle_interrupt(interrupt_tx));
+        let app = Self { config, db, interrupt: interrupt_rx };
         app.bitcoind()?
             .get_blockchain_info()
             .context("Failed to connect to bitcoind. Check config RPC connection.")?;
@@ -116,7 +120,13 @@ impl AppTrait for App {
         );
         println!("{}", pj_uri_string);
 
-        self.start_http_server().await?;
+        let mut interrupt = self.interrupt.clone();
+        tokio::select! {
+            res = self.start_http_server() => { res?; }
+            _ = interrupt.changed() => {
+                println!("Interrupted.");
+            }
+        }
         Ok(())
     }
 }

payjoin-cli/src/app/v2.rs

@@ -10,12 +10,11 @@ use payjoin::receive::v2::{Receiver, UncheckedProposal};
 use payjoin::receive::Error;
 use payjoin::send::v2::{Sender, SenderBuilder};
 use payjoin::{bitcoin, Uri};
-use tokio::signal;
 use tokio::sync::watch;
 
 use super::config::AppConfig;
 use super::App as AppTrait;
-use crate::app::{http_agent, input_pair_from_list_unspent};
+use crate::app::{handle_interrupt, http_agent, input_pair_from_list_unspent};
 use crate::db::Database;
 
 #[derive(Clone)]
@@ -399,13 +398,6 @@ async fn unwrap_ohttp_keys_or_else_fetch(config: &AppConfig) -> Result<payjoin::
     }
 }
 
-async fn handle_interrupt(tx: watch::Sender<()>) {
-    if let Err(e) = signal::ctrl_c().await {
-        eprintln!("Error setting up Ctrl-C handler: {}", e);
-    }
-    let _ = tx.send(());
-}
-
 async fn post_request(req: payjoin::Request) -> Result<reqwest::Response> {
     let http = http_agent()?;
     http.post(req.url)

payjoin-cli/tests/e2e.rs

@@ -1,7 +1,7 @@
 #[cfg(feature = "_danger-local-https")]
 mod e2e {
     use std::env;
-    use std::process::Stdio;
+    use std::process::{ExitStatus, Stdio};
 
     use nix::sys::signal::{kill, Signal};
     use nix::unistd::Pid;
@@ -10,9 +10,11 @@ mod e2e {
     use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
     use tokio::process::Command;
 
-    fn sigint(child: &tokio::process::Child) -> nix::Result<()> {
+    async fn terminate(mut child: tokio::process::Child) -> tokio::io::Result<ExitStatus> {
         let pid = child.id().expect("Failed to get child PID");
-        kill(Pid::from_raw(pid as i32), Signal::SIGINT)
+        kill(Pid::from_raw(pid as i32), Signal::SIGINT)?;
+        // wait for child process to exit completely
+        child.wait().await
     }
 
     const RECEIVE_SATS: &str = "54321";
@@ -119,8 +121,9 @@ mod e2e {
                 .unwrap_or(Some(false)) // timed out
                 .expect("rx channel closed prematurely"); // recv() returned None
 
-            sigint(&cli_receiver).expect("Failed to kill payjoin-cli");
-            sigint(&cli_sender).expect("Failed to kill payjoin-cli");
+            terminate(cli_receiver).await.expect("Failed to kill payjoin-cli");
+            terminate(cli_sender).await.expect("Failed to kill payjoin-cli");
+
             payjoin_sent
         })
         .await?;
@@ -143,7 +146,7 @@ mod e2e {
         use std::path::PathBuf;
 
         use payjoin_test_utils::{init_tracing, TestServices};
-        use tokio::process::Child;
+        use tokio::process::{Child, ChildStdout};
 
         type Result<T> = std::result::Result<T, BoxError>;
 
@@ -263,126 +266,94 @@ mod e2e {
         }
 
         async fn get_bip21_from_receiver(mut cli_receiver: Child) -> String {
-            let stdout =
-                cli_receiver.stdout.take().expect("Failed to take stdout of child process");
-            let reader = BufReader::new(stdout);
-            let mut stdout = tokio::io::stdout();
-            let mut bip21 = String::new();
-
-            let mut lines = reader.lines();
-
-            while let Some(line) = lines.next_line().await.expect("Failed to read line from stdout")
-            {
-                // Write to stdout regardless
-                stdout
-                    .write_all(format!("{}\n", line).as_bytes())
-                    .await
-                    .expect("Failed to write to stdout");
-
-                if line.to_ascii_uppercase().starts_with("BITCOIN") {
-                    bip21 = line;
-                    break;
-                }
-            }
+            let mut stdout =
+                cli_receiver.stdout.take().expect("failed to take stdout of child process");
+            let bip21 = wait_for_stdout_match(&mut stdout, |line| {
+                line.to_ascii_uppercase().starts_with("BITCOIN")
+            })
+            .await
+            .expect("payjoin-cli receiver should output a bitcoin URI");
             log::debug!("Got bip21 {}", &bip21);
 
-            sigint(&cli_receiver).expect("Failed to kill payjoin-cli");
+            terminate(cli_receiver).await.expect("Failed to kill payjoin-cli");
             bip21
         }
 
         async fn send_until_request_timeout(mut cli_sender: Child) -> Result<()> {
-            let stdout = cli_sender.stdout.take().expect("Failed to take stdout of child process");
-            let reader = BufReader::new(stdout);
-            let (tx, mut rx) = tokio::sync::mpsc::channel(1);
-
-            let mut lines = reader.lines();
-            tokio::spawn(async move {
-                let mut stdout = tokio::io::stdout();
-                while let Some(line) =
-                    lines.next_line().await.expect("Failed to read line from stdout")
-                {
-                    stdout
-                        .write_all(format!("{}\n", line).as_bytes())
-                        .await
-                        .expect("Failed to write to stdout");
-                    if line.contains("No response yet.") {
-                        let _ = tx.send(true).await;
-                        break;
-                    }
-                }
-            });
-
+            let mut stdout =
+                cli_sender.stdout.take().expect("failed to take stdout of child process");
             let timeout = tokio::time::Duration::from_secs(35);
-            let fallback_sent = tokio::time::timeout(timeout, rx.recv()).await?;
-
-            sigint(&cli_sender).expect("Failed to kill payjoin-cli initial sender");
-
-            assert!(fallback_sent.unwrap_or(false), "Fallback send was not detected");
+            let res = tokio::time::timeout(
+                timeout,
+                wait_for_stdout_match(&mut stdout, |line| line.contains("No response yet.")),
+            )
+            .await?;
+
+            terminate(cli_sender).await.expect("Failed to kill payjoin-cli initial sender");
+            assert!(res.is_some(), "Fallback send was not detected");
             Ok(())
         }
 
         async fn respond_with_payjoin(mut cli_receive_resumer: Child) -> Result<()> {
-            let stdout =
+            let mut stdout =
                 cli_receive_resumer.stdout.take().expect("Failed to take stdout of child process");
-            let reader = BufReader::new(stdout);
-            let (tx, mut rx) = tokio::sync::mpsc::channel(1);
-
-            let mut lines = reader.lines();
-            tokio::spawn(async move {
-                let mut stdout = tokio::io::stdout();
-                while let Some(line) =
-                    lines.next_line().await.expect("Failed to read line from stdout")
-                {
-                    stdout
-                        .write_all(format!("{}\n", line).as_bytes())
-                        .await
-                        .expect("Failed to write to stdout");
-                    if line.contains("Response successful") {
-                        let _ = tx.send(true).await;
-                        break;
-                    }
-                }
-            });
-
             let timeout = tokio::time::Duration::from_secs(10);
-            let response_successful = tokio::time::timeout(timeout, rx.recv()).await?;
-
-            sigint(&cli_receive_resumer).expect("Failed to kill payjoin-cli");
-
-            assert!(response_successful.unwrap_or(false), "Did not respond with Payjoin PSBT");
+            let res = tokio::time::timeout(
+                timeout,
+                wait_for_stdout_match(&mut stdout, |line| line.contains("Response successful")),
+            )
+            .await?;
+
+            terminate(cli_receive_resumer).await.expect("Failed to kill payjoin-cli");
+            assert!(res.is_some(), "Did not respond with Payjoin PSBT");
             Ok(())
         }
 
         async fn check_payjoin_sent(mut cli_send_resumer: Child) -> Result<()> {
-            let stdout =
+            let mut stdout =
                 cli_send_resumer.stdout.take().expect("Failed to take stdout of child process");
-            let reader = BufReader::new(stdout);
-            let (tx, mut rx) = tokio::sync::mpsc::channel(1);
+            let timeout = tokio::time::Duration::from_secs(10);
+            let res = tokio::time::timeout(
+                timeout,
+                wait_for_stdout_match(&mut stdout, |line| line.contains("Payjoin sent")),
+            )
+            .await?;
+
+            terminate(cli_send_resumer).await.expect("Failed to kill payjoin-cli");
+            assert!(res.is_some(), "Payjoin send was not detected");
+            Ok(())
+        }
 
+        /// Read lines from `child_stdout` until `match_pattern` is found and the corresponding
+        /// line is returned.
+        /// Also writes every read line to tokio::io::stdout();
+        async fn wait_for_stdout_match<F>(
+            child_stdout: &mut ChildStdout,
+            match_pattern: F,
+        ) -> Option<String>
+        where
+            F: Fn(&str) -> bool,
+        {
+            let reader = BufReader::new(child_stdout);
             let mut lines = reader.lines();
-            tokio::spawn(async move {
-                let mut stdout = tokio::io::stdout();
-                while let Some(line) =
-                    lines.next_line().await.expect("Failed to read line from stdout")
-                {
-                    stdout
-                        .write_all(format!("{}\n", line).as_bytes())
-                        .await
-                        .expect("Failed to write to stdout");
-                    if line.contains("Payjoin sent") {
-                        let _ = tx.send(true).await;
-                        break;
-                    }
-                }
-            });
+            let mut res = None;
 
-            let timeout = tokio::time::Duration::from_secs(10);
-            let payjoin_sent = tokio::time::timeout(timeout, rx.recv()).await?;
+            let mut stdout = tokio::io::stdout();
+            while let Some(line) = lines.next_line().await.expect("Failed to read line from stdout")
+            {
+                // Write all output to tests stdout
+                stdout
+                    .write_all(format!("{}\n", line).as_bytes())
+                    .await
+                    .expect("Failed to write to stdout");
 
-            sigint(&cli_send_resumer).expect("Failed to kill payjoin-cli");
+                if match_pattern(&line) {
+                    res = Some(line);
+                    break;
+                }
+            }
 
-            assert!(payjoin_sent.unwrap_or(false), "Payjoin send was not detected");
-            Ok(())
+            res
         }
 
         Ok(())