test(app): keep support package-local and bound client runs

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();

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<oneshot::Sender<()>>,
+    done: Option<JoinHandle<Result<(), irtt_server::ServerRuntimeError>>>,
+}
+
+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();
+    }
+}