From ae699c4ef21c866061d12096bbc018f5b0f53421 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 5 May 2026 00:58:18 +0700 Subject: [PATCH] test: refactor real backend support for reliable output capture --- crates/irtt-client/tests/support/mod.rs | 16 +- crates/irtt-client/tests/support/real_irtt.rs | 167 ++++++++---------- 2 files changed, 90 insertions(+), 93 deletions(-) diff --git a/crates/irtt-client/tests/support/mod.rs b/crates/irtt-client/tests/support/mod.rs index 16d1f3b..42715a3 100644 --- a/crates/irtt-client/tests/support/mod.rs +++ b/crates/irtt-client/tests/support/mod.rs @@ -36,15 +36,27 @@ pub fn selected_backend() -> TestBackendKind { Ok("real") => { let irtt_bin = std::env::var("IRTT_BIN").unwrap_or_else(|_| "irtt".to_string()); match Command::new(&irtt_bin).arg("version").output() { - Ok(output) if output.status.success() => TestBackendKind::Real, + Ok(output) if output.status.success() => { + debug_backend("[backend] selected backend=real"); + TestBackendKind::Real + } _ => panic!("IRTT_TEST_BACKEND=real but irtt binary not found at '{irtt_bin}'"), } } - Ok("fake") | Err(_) => TestBackendKind::Fake, + Ok("fake") | Err(_) => { + debug_backend("[backend] selected backend=fake"); + TestBackendKind::Fake + } Ok(other) => panic!("unknown IRTT_TEST_BACKEND value: {other}"), } } +fn debug_backend(message: &str) { + if std::env::var("IRTT_TEST_BACKEND_DEBUG").as_deref() == Ok("1") { + eprintln!("{message}"); + } +} + pub enum BackendPeer { Fake(FakeServer), Real(RealIrtServer), diff --git a/crates/irtt-client/tests/support/real_irtt.rs b/crates/irtt-client/tests/support/real_irtt.rs index 0c09313..2f8f0d3 100644 --- a/crates/irtt-client/tests/support/real_irtt.rs +++ b/crates/irtt-client/tests/support/real_irtt.rs @@ -1,31 +1,32 @@ #![allow(dead_code)] -use std::io::{BufRead, BufReader}; +use std::io::{BufRead, BufReader, Read}; use std::net::{SocketAddr, UdpSocket}; use std::process::{Child, Command, Stdio}; -use std::sync::mpsc; -use std::thread; +use std::sync::{mpsc, Arc, Mutex}; +use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; pub struct RealIrtServer { addr: SocketAddr, child: Child, keepalive_ms: Option, + stdout_thread: Option>, + stderr_thread: Option>, } impl RealIrtServer { pub fn start(hmac_key: Option<&[u8]>) -> Result { let port = Self::find_free_port()?; let addr = SocketAddr::from(([127, 0, 0, 1], port)); - let bind = format!("127.0.0.1:{}", port); + let bind = format!("127.0.0.1:{port}"); let irtt_bin = std::env::var("IRTT_BIN").unwrap_or_else(|_| "irtt".to_string()); - let keepalive_ms = std::env::var("IRTT_TEST_KEEP_SERVER_MS") .ok() .and_then(|v| v.parse::().ok()); - debug_eprintln!("[real_irtt] backend=real irtt_bin={irtt_bin} bind={bind}"); + debug_eprintln!("[real_irtt] selected backend=real irtt_bin={irtt_bin} bind={bind}"); let mut cmd = Command::new(&irtt_bin); cmd.arg("server") @@ -37,7 +38,7 @@ impl RealIrtServer { if let Some(key) = hmac_key { let hex_key: String = key.iter().map(|b| format!("{:02x}", b)).collect(); - cmd.arg(format!("--hmac=0x{}", hex_key)); + cmd.arg(format!("--hmac=0x{hex_key}")); } let full_cmd = format!("{cmd:?}"); @@ -45,7 +46,7 @@ impl RealIrtServer { let mut child = cmd .spawn() - .map_err(|e| format!("failed to spawn irtt server: {}", e))?; + .map_err(|e| format!("failed to spawn irtt server: {e}"))?; let pid = child.id(); debug_eprintln!("[real_irtt] child pid={pid}"); @@ -53,91 +54,49 @@ impl RealIrtServer { let stdout = child.stdout.take().ok_or("failed to capture stdout")?; let stderr = child.stderr.take().ok_or("failed to capture stderr")?; - let (out_tx, out_rx) = mpsc::channel(); - let (err_tx, err_rx) = mpsc::channel(); + let captured = Arc::new(Mutex::new(Vec::new())); + let (line_tx, line_rx) = mpsc::channel(); - thread::spawn(move || { - let reader = BufReader::new(stdout); - for line in reader.lines() { - match line { - Ok(l) => { - if out_tx.send(("stdout", l)).is_err() { - break; - } - } - Err(_) => break, - } - } - }); - - thread::spawn(move || { - let reader = BufReader::new(stderr); - for line in reader.lines() { - match line { - Ok(l) => { - if err_tx.send(("stderr", l)).is_err() { - break; - } - } - Err(_) => break, - } - } - }); - - let timeout = Duration::from_secs(5); - let start = Instant::now(); - let mut captured_lines = Vec::new(); - - let ready = loop { - let remaining = timeout - .checked_sub(start.elapsed()) - .unwrap_or(Duration::ZERO); - - let line = select_line(&out_rx, &err_rx, remaining); - match line { - Some((stream, text)) => { - debug_eprintln!("[real_irtt] {stream}: {text}"); - captured_lines.push(format!("{stream}: {text}")); - if text.contains("[ListenerStart]") { - break true; - } - if (text.contains("[ListenerStop]") || text.contains("[ServerStop]")) - && (text.contains("error") || text.contains("Error")) - { - break false; - } - } - None => { - break false; - } - } + let stdout_thread = { + let line_tx = line_tx.clone(); + let captured = Arc::clone(&captured); + thread::spawn(move || drain_stream("stdout", stdout, line_tx, captured)) }; + let stderr_thread = { + let captured = Arc::clone(&captured); + thread::spawn(move || drain_stream("stderr", stderr, line_tx, captured)) + }; + + let ready = wait_for_ready(&line_rx, Duration::from_secs(5)); + if !ready { let exit_status = child.try_wait().ok().flatten(); - debug_eprintln!("[real_irtt] startup failed, exit_status={exit_status:?}"); + debug_eprintln!("[real_irtt] readiness result=false exit_status={exit_status:?}"); child.kill().ok(); child.wait().ok(); + stdout_thread.join().ok(); + stderr_thread.join().ok(); - let output = captured_lines.join("\n"); + let output = captured.lock().unwrap().join("\n"); let extra = match exit_status { Some(status) => format!("\nexit status: {status}"), None => String::new(), }; + debug_eprintln!("[real_irtt] failure diagnostics:\n{output}{extra}"); return Err(format!( "irtt server failed to start\noutput:\n{output}{extra}" )); } - debug_eprintln!("[real_irtt] ready on {addr}"); - - drop(out_rx); - drop(err_rx); + debug_eprintln!("[real_irtt] readiness result=true addr={addr}"); Ok(Self { addr, child, keepalive_ms, + stdout_thread: Some(stdout_thread), + stderr_thread: Some(stderr_thread), }) } @@ -146,11 +105,11 @@ impl RealIrtServer { } fn find_free_port() -> Result { - let socket = UdpSocket::bind("127.0.0.1:0") - .map_err(|e| format!("failed to find free port: {}", e))?; + let socket = + UdpSocket::bind("127.0.0.1:0").map_err(|e| format!("failed to find free port: {e}"))?; let port = socket .local_addr() - .map_err(|e| format!("failed to get local addr: {}", e))? + .map_err(|e| format!("failed to get local addr: {e}"))? .port(); drop(socket); Ok(port) @@ -169,29 +128,55 @@ impl Drop for RealIrtServer { debug_eprintln!("[real_irtt] killing child pid={}", self.child.id()); self.child.kill().ok(); self.child.wait().ok(); + if let Some(thread) = self.stdout_thread.take() { + thread.join().ok(); + } + if let Some(thread) = self.stderr_thread.take() { + thread.join().ok(); + } } } -fn select_line( - out_rx: &mpsc::Receiver<(&str, String)>, - err_rx: &mpsc::Receiver<(&str, String)>, - timeout: Duration, -) -> Option<(&'static str, String)> { - if let Ok((_, line)) = out_rx.recv_timeout(timeout) { - return Some(("stdout", line)); - } - if let Ok((_, line)) = err_rx.recv_timeout(Duration::ZERO) { - return Some(("stderr", line)); +fn wait_for_ready(line_rx: &mpsc::Receiver<(&'static str, String)>, timeout: Duration) -> bool { + let start = Instant::now(); + loop { + let remaining = timeout + .checked_sub(start.elapsed()) + .unwrap_or(Duration::ZERO); + + match line_rx.recv_timeout(remaining) { + Ok((stream, text)) => { + debug_eprintln!("[real_irtt] {stream}: {text}"); + if text.contains("[ListenerStart]") { + return true; + } + if (text.contains("[ListenerStop]") || text.contains("[ServerStop]")) + && (text.contains("error") || text.contains("Error")) + { + return false; + } + } + Err(_) => return false, + } } - if timeout > Duration::ZERO { - if let Ok((_, line)) = out_rx.recv_timeout(Duration::from_millis(100)) { - return Some(("stdout", line)); - } - if let Ok((_, line)) = err_rx.recv_timeout(Duration::from_millis(100)) { - return Some(("stderr", line)); +} + +fn drain_stream( + stream: &'static str, + pipe: impl Read, + tx: mpsc::Sender<(&'static str, String)>, + captured: Arc>>, +) { + let reader = BufReader::new(pipe); + for line in reader.lines() { + match line { + Ok(text) => { + captured.lock().unwrap().push(format!("{stream}: {text}")); + let _ = tx.send((stream, text)); + } + Err(_) => break, } } - None } macro_rules! debug_eprintln {