diff --git a/crates/irtt-app/tests/cli.rs b/crates/irtt-app/tests/cli.rs index 206662e..1c46d5a 100644 --- a/crates/irtt-app/tests/cli.rs +++ b/crates/irtt-app/tests/cli.rs @@ -1,11 +1,39 @@ #[cfg(feature = "client")] -#[path = "../../irtt-integration-tests/tests/support/in_tree_server.rs"] +#[path = "support/in_tree_server.rs"] mod in_tree_server; #[cfg(feature = "client")] use irtt_server::ServerConfig; use std::process::Command; +#[cfg(feature = "client")] +struct ClientProcess(std::process::Child); + +#[cfg(feature = "client")] +impl Drop for ClientProcess { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[cfg(feature = "client")] +impl ClientProcess { + fn wait(&mut self) -> std::process::ExitStatus { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + if let Some(status) = self.0.try_wait().unwrap() { + return status; + } + assert!( + std::time::Instant::now() < deadline, + "client did not terminate within five seconds" + ); + std::thread::sleep(std::time::Duration::from_millis(10)); + } + } +} + // Dispatch/help/availability belong to the application and run in every build. #[test] fn dispatcher_reports_and_runs_the_enabled_applets() { @@ -63,6 +91,7 @@ fn dispatcher_reports_and_runs_the_enabled_applets() { #[cfg(feature = "client")] #[test] fn client_arguments_export_a_real_client_run() { + use std::{io::Read, process::Stdio, thread}; let dispatcher = env!("CARGO_BIN_EXE_irtt-rs"); for option in ["--dscp", "--ttl"] { assert!(!Command::new(dispatcher) @@ -74,41 +103,58 @@ fn client_arguments_export_a_real_client_run() { } let server = in_tree_server::InTreeServer::start(ServerConfig::default().with_hmac_key(b"cli-test")); - let output = Command::new(dispatcher) - .args([ - "client", - "--duration", - "100ms", - "--interval", - "20ms", - "--hmac", - "cli-test", - "--length", - "128", - "--sfill", - "pattern:ab", - "--clock", - "wall", - "--tstamp", - "both", - "--stats", - "both", - "--ttl", - "64", - "--format", - "jsonl", - "--columns", - "target,event,bytes", - &format!("edge={}", server.addr), - ]) - .output() - .unwrap(); - assert!( - output.status.success(), - "{}", - String::from_utf8_lossy(&output.stderr) + let mut client = ClientProcess( + Command::new(dispatcher) + .args([ + "client", + "--duration", + "100ms", + "--interval", + "20ms", + "--hmac", + "cli-test", + "--length", + "128", + "--sfill", + "pattern:ab", + "--clock", + "wall", + "--tstamp", + "both", + "--stats", + "both", + "--ttl", + "64", + "--format", + "jsonl", + "--columns", + "target,event,bytes", + &format!("edge={}", server.addr), + ]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), ); - let stdout = String::from_utf8(output.stdout).unwrap(); + // Drain both pipes while waiting so output cannot block client termination. + let mut stdout = client.0.stdout.take().unwrap(); + let stdout = thread::spawn(move || { + let mut bytes = Vec::new(); + stdout.read_to_end(&mut bytes).unwrap(); + bytes + }); + let mut stderr = client.0.stderr.take().unwrap(); + let stderr = thread::spawn(move || { + let mut bytes = Vec::new(); + stderr.read_to_end(&mut bytes).unwrap(); + bytes + }); + let status = client.wait(); + let stdout = stdout.join().unwrap(); + let stderr = stderr.join().unwrap(); + assert!(status.success(), "{}", String::from_utf8_lossy(&stderr)); + let stdout = String::from_utf8(stdout).unwrap(); let rows: Vec<_> = stdout.lines().collect(); assert!(!rows.is_empty()); assert!(rows.iter().all(|row| row.starts_with('{') @@ -263,18 +309,11 @@ fn server_command_line_binds_replace_the_environment_bind() { fn targets_stdin_supersedes_the_active_set_and_stops_on_eof() { use std::{ io::{BufRead, BufReader, Read, Write}, - process::{Child, Stdio}, + process::Stdio, sync::mpsc, thread, time::{Duration, Instant}, }; - struct ClientProcess(Child); - impl Drop for ClientProcess { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); - } - } let server = in_tree_server::InTreeServer::start(ServerConfig::default()); let mut client = ClientProcess( Command::new(env!("CARGO_BIN_EXE_irtt-client")) @@ -334,17 +373,7 @@ fn targets_stdin_supersedes_the_active_set_and_stops_on_eof() { ); } drop(stdin); - let deadline = Instant::now() + Duration::from_secs(5); - let status = loop { - if let Some(status) = client.0.try_wait().unwrap() { - break status; - } - assert!( - Instant::now() < deadline, - "client did not terminate on stdin EOF" - ); - thread::sleep(Duration::from_millis(10)); - }; + let status = client.wait(); reader.join().unwrap(); rows.extend(rx.try_iter()); let mut stderr = String::new(); diff --git a/crates/irtt-app/tests/support/in_tree_server.rs b/crates/irtt-app/tests/support/in_tree_server.rs new file mode 100644 index 0000000..faf4fcd --- /dev/null +++ b/crates/irtt-app/tests/support/in_tree_server.rs @@ -0,0 +1,74 @@ +use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::mpsc, + thread::{self, JoinHandle}, + time::Duration, +}; + +use irtt_server::{Server, ServerConfig}; +use tokio::sync::oneshot; + +const STARTUP_TIMEOUT: Duration = Duration::from_secs(2); + +pub struct InTreeServer { + pub addr: SocketAddr, + shutdown: Option>, + done: Option>>, +} + +impl InTreeServer { + pub fn start(config: ServerConfig) -> Self { + let (startup_tx, startup_rx) = mpsc::sync_channel(1); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let done = thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("failed to build in-tree test server runtime"); + runtime.block_on(async move { + let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + let mut server = Server::bind(bind_addr, config).await?; + startup_tx + .send(server.local_addr()?) + .expect("test stopped waiting for in-tree server startup"); + server + .run(async { + let _ = shutdown_rx.await; + }) + .await + }) + }); + + let addr = match startup_rx.recv_timeout(STARTUP_TIMEOUT) { + Ok(addr) => addr, + Err(error) => { + let _ = shutdown_tx.send(()); + let result = done.join().expect("in-tree test server thread panicked"); + panic!("in-tree test server failed before startup ({error}): {result:?}"); + } + }; + + Self { + addr, + shutdown: Some(shutdown_tx), + done: Some(done), + } + } + + fn stop_inner(&mut self) { + if let Some(shutdown) = self.shutdown.take() { + let _ = shutdown.send(()); + } + if let Some(done) = self.done.take() { + done.join() + .expect("in-tree test server thread panicked") + .expect("in-tree test server stopped with an error"); + } + } +} + +impl Drop for InTreeServer { + fn drop(&mut self) { + self.stop_inner(); + } +}