Lochnair-Labs/irtt-rs
2
0 Fork 0
Code Issues 1 Pull Requests 4 Actions 4 Packages Projects Releases 6 Wiki Activity

test: refactor real backend support for reliable output capture

Browse Source
This commit is contained in:
Nils Andreas Svee committed 2026-05-05 00:58:22 +07:00
1 parent 8625672f85
commit ae699c4ef2
2 files changed
+90 -93
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/irtt-client/tests/support/mod.rs
+14 -2
View File
@@ -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),
crates/irtt-client/tests/support/real_irtt.rs
+76 -91
View File
@@ -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<u64>,
stdout_thread: Option<JoinHandle<()>>,
stderr_thread: Option<JoinHandle<()>>,
}
impl RealIrtServer {
pub fn start(hmac_key: Option<&[u8]>) -> Result<Self, String> {
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::<u64>().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<u16, String> {
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<Mutex<Vec<String>>>,
) {
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 {