aboutsummaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/client_mode.rs303
-rw-r--r--tests/detach_reattach.rs373
-rw-r--r--tests/support/mod.rs267
3 files changed, 371 insertions, 572 deletions
diff --git a/tests/client_mode.rs b/tests/client_mode.rs
index 882bcb1c..8860f65d 100644
--- a/tests/client_mode.rs
+++ b/tests/client_mode.rs
@@ -12,8 +12,10 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize};
use support::{
- cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid,
- unregister_spawned_herdr_pid,
+ cleanup_test_base, client_handshake, encode_varint_u16, encode_varint_u32, frame_message,
+ read_server_message, register_runtime_dir, register_spawned_herdr_pid,
+ unregister_spawned_herdr_pid, wait_for_file, wait_for_message_variant, wait_for_socket,
+ wait_until,
};
fn unique_test_dir() -> PathBuf {
@@ -66,28 +68,6 @@ fn test_lock() -> MutexGuard<'static, ()> {
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
-fn wait_for_socket(path: &PathBuf, timeout: Duration) {
- let deadline = Instant::now() + timeout;
- while Instant::now() < deadline {
- if path.exists() && UnixStream::connect(path).is_ok() {
- return;
- }
- thread::sleep(Duration::from_millis(25));
- }
- panic!("socket did not appear at {}", path.display());
-}
-
-fn wait_for_file(path: &PathBuf, timeout: Duration) {
- let deadline = Instant::now() + timeout;
- while Instant::now() < deadline {
- if path.exists() {
- return;
- }
- thread::sleep(Duration::from_millis(25));
- }
- panic!("file did not appear at {}", path.display());
-}
-
fn spawn_client_process(
config_home: &PathBuf,
runtime_dir: &PathBuf,
@@ -178,203 +158,6 @@ fn ping_socket(socket_path: &PathBuf) -> String {
}
// ---------------------------------------------------------------------------
-// Bincode v2 varint encoding helpers (same as server_headless.rs tests)
-// ---------------------------------------------------------------------------
-
-/// Encode a varint u32 value according to bincode v2 VarintEncoding.
-fn encode_varint_u32(v: u32) -> Vec<u8> {
- if v < 251 {
- vec![v as u8]
- } else if v < 65536 {
- let mut buf = vec![251u8];
- buf.extend_from_slice(&(v as u16).to_le_bytes());
- buf
- } else {
- let mut buf = vec![252u8];
- buf.extend_from_slice(&v.to_le_bytes());
- buf
- }
-}
-
-/// Encode a varint u16 value.
-fn encode_varint_u16(v: u16) -> Vec<u8> {
- if v < 251 {
- vec![v as u8]
- } else {
- let mut buf = vec![251u8];
- buf.extend_from_slice(&v.to_le_bytes());
- buf
- }
-}
-
-/// Encode an enum variant with its fields.
-fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> {
- let mut buf = encode_varint_u32(variant_idx);
- for field in fields {
- buf.extend_from_slice(field);
- }
- buf
-}
-
-/// Frame a message with u32LE length prefix.
-fn frame_message(payload: &[u8]) -> Vec<u8> {
- let len = payload.len() as u32;
- let mut framed = len.to_le_bytes().to_vec();
- framed.extend_from_slice(payload);
- framed
-}
-
-/// Decode a varint u32 from a byte slice at the given offset.
-fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> {
- if offset >= payload.len() {
- return Err("payload too short for varint".into());
- }
- let first_byte = payload[offset];
- match first_byte {
- 0..=250 => Ok((first_byte as u32, 1)),
- 251 => {
- if offset + 3 > payload.len() {
- return Err("payload too short for u16 varint".into());
- }
- let v = u16::from_le_bytes(
- payload[offset + 1..offset + 3]
- .try_into()
- .map_err(|e: std::array::TryFromSliceError| e.to_string())?,
- );
- Ok((v as u32, 3))
- }
- 252 => {
- if offset + 5 > payload.len() {
- return Err("payload too short for u32 varint".into());
- }
- let v = u32::from_le_bytes(
- payload[offset + 1..offset + 5]
- .try_into()
- .map_err(|e: std::array::TryFromSliceError| e.to_string())?,
- );
- Ok((v, 5))
- }
- _ => Err(format!("unsupported varint tag: {first_byte}")),
- }
-}
-
-/// Sends a Hello message over the client socket and reads the Welcome response.
-fn client_handshake(
- stream: &mut UnixStream,
- version: u32,
- cols: u16,
- rows: u16,
-) -> Result<(u32, Option<String>), String> {
- stream
- .set_read_timeout(Some(Duration::from_secs(5)))
- .map_err(|e| e.to_string())?;
-
- // Encode Hello message using bincode v2 varint format.
- // ClientMessage::Hello is variant 0.
- let hello_payload = encode_varint_enum(
- 0,
- &[
- &encode_varint_u32(version),
- &encode_varint_u16(cols),
- &encode_varint_u16(rows),
- ],
- );
- let framed = frame_message(&hello_payload);
- stream.write_all(&framed).map_err(|e| e.to_string())?;
- stream.flush().map_err(|e| e.to_string())?;
-
- // Read the framed response.
- let mut len_buf = [0u8; 4];
- stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?;
- let len = u32::from_le_bytes(len_buf) as usize;
-
- if len > 2 * 1024 * 1024 {
- return Err(format!("oversized response: {len}"));
- }
-
- let mut payload = vec![0u8; len];
- stream.read_exact(&mut payload).map_err(|e| e.to_string())?;
-
- // Decode Welcome: ServerMessage variant 0 = Welcome { version: u32, error: Option<String> }
- decode_welcome(&payload)
-}
-
-/// Decode a ServerMessage::Welcome from bincode v2 payload.
-fn decode_welcome(payload: &[u8]) -> Result<(u32, Option<String>), String> {
- let mut offset = 0;
-
- // Variant index (should be 0 for Welcome)
- let (variant, consumed) = decode_varint_u32(payload, offset)?;
- offset += consumed;
- if variant != 0 {
- return Err(format!(
- "expected Welcome (variant 0), got variant {variant}"
- ));
- }
-
- // version: u32
- let (version, consumed) = decode_varint_u32(payload, offset)?;
- offset += consumed;
-
- // error: Option<String> — discriminant is always 1 byte
- if offset >= payload.len() {
- return Err("payload too short for Option tag".into());
- }
- let option_tag = payload[offset];
- offset += 1;
-
- let error = if option_tag == 1 {
- // Some(String) — length as varint + UTF-8 bytes
- let (str_len, consumed) = decode_varint_u32(payload, offset)?;
- offset += consumed;
- let str_len = str_len as usize;
-
- if offset + str_len > payload.len() {
- return Err("payload too short for string content".into());
- }
- let s = String::from_utf8(payload[offset..offset + str_len].to_vec())
- .map_err(|e| e.to_string())?;
- Some(s)
- } else {
- None
- };
-
- Ok((version, error))
-}
-
-/// Read a framed ServerMessage from the stream, returning the variant index
-/// and the raw payload for further decoding.
-fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec<u8>), String> {
- stream
- .set_read_timeout(Some(Duration::from_secs(10)))
- .map_err(|e| e.to_string())?;
-
- let mut len_buf = [0u8; 4];
- stream
- .read_exact(&mut len_buf)
- .map_err(|e| format!("read length prefix: {e}"))?;
- let len = u32::from_le_bytes(len_buf) as usize;
-
- if len > 2 * 1024 * 1024 {
- return Err(format!("oversized frame: {len} bytes"));
- }
- if len == 0 {
- return Err("zero-length frame".into());
- }
-
- let mut payload = vec![0u8; len];
- stream
- .read_exact(&mut payload)
- .map_err(|e| format!("read payload: {e}"))?;
-
- // Decode the variant index.
- let (variant, consumed) = decode_varint_u32(&payload, 0)?;
-
- // Return the variant index and the remaining payload (after the variant tag).
- Ok((variant, payload[consumed..].to_vec()))
-}
-
-// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
@@ -405,10 +188,6 @@ fn client_connects_and_receives_frame() {
error
);
- // Give the server a moment to set up the writer thread and
- // start the render cycle after learning our terminal size.
- thread::sleep(Duration::from_millis(500));
-
// Read the next message from the server — should be a Frame (variant 1).
stream.set_nonblocking(false).unwrap();
stream
@@ -465,8 +244,12 @@ fn client_input_forwarded_to_pane() {
.expect("should send Input message");
stream.flush().expect("should flush");
- // Give the server time to process the input and render.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should still respond to ping after input"
+ );
// Verify the server is still alive and responsive via API.
let response = ping_socket(&api_socket);
@@ -518,8 +301,12 @@ fn client_resize_sends_message() {
.expect("should send Resize message");
stream.flush().expect("should flush");
- // Give the server time to process and render at the new size.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should respond after resize"
+ );
// Verify the server is still alive.
let response = ping_socket(&api_socket);
@@ -676,11 +463,7 @@ fn server_crash_after_attach_causes_lost_connection_error() {
match thin_reader.read(&mut buf) {
Ok(n) if n > 0 => {
let out = String::from_utf8_lossy(&buf[..n]);
- if out.contains("\u{2500}")
- || out.contains("workspace")
- || out.contains("terminal")
- || out.contains("pane")
- {
+ if !out.is_empty() {
seen = true;
break;
}
@@ -720,7 +503,7 @@ fn server_crash_after_attach_causes_lost_connection_error() {
crash_output.push_str(&String::from_utf8_lossy(&buf[..n]));
}
}
- thread::sleep(Duration::from_millis(50));
+ thread::sleep(Duration::from_millis(20));
}
exited
};
@@ -805,26 +588,10 @@ fn client_receives_frame_after_pane_output() {
stream.write_all(&framed).expect("send input");
stream.flush().expect("flush");
- // Wait for the server to process and render.
- thread::sleep(Duration::from_millis(500));
-
// Read subsequent frames — the server should have re-rendered after
// the input was processed.
- stream
- .set_read_timeout(Some(Duration::from_secs(5)))
- .unwrap();
- let mut received_frame = false;
- for _ in 0..5 {
- match read_server_message(&mut stream) {
- Ok((variant, _)) => {
- if variant == 1 {
- received_frame = true;
- break;
- }
- }
- Err(_) => break,
- }
- }
+ let received_frame = wait_for_message_variant(&mut stream, Duration::from_secs(2), 1)
+ .expect("wait for post-output frame");
assert!(received_frame, "should receive a Frame after pane output");
cleanup_spawned_herdr(spawned, base);
@@ -872,8 +639,11 @@ fn navigate_mode_keybind_dispatch_in_server() {
stream.write_all(&framed).expect("send prefix key");
stream.flush().expect("flush");
- // Give the server time to process.
- thread::sleep(Duration::from_millis(100));
+ stream
+ .set_read_timeout(Some(Duration::from_millis(200)))
+ .unwrap();
+ while let Ok(_) = read_server_message(&mut stream) {}
+ stream.set_read_timeout(None).unwrap();
// Send 'n' (new workspace in navigate mode).
let n_input = b"n".to_vec();
@@ -887,8 +657,12 @@ fn navigate_mode_keybind_dispatch_in_server() {
stream.write_all(&framed).expect("send n key");
stream.flush().expect("flush");
- // Wait for processing and render.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should still respond after navigate mode input"
+ );
// Verify the server is still alive and the API still works.
let response = ping_socket(&api_socket);
@@ -1163,8 +937,12 @@ fn client_receives_notify_on_agent_state_change() {
let mut focus_response = String::new();
focus_reader.read_line(&mut focus_response).unwrap();
- // Give the server a moment to process.
- thread::sleep(Duration::from_millis(100));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should stay responsive after workspace focus"
+ );
// Report agent as Working first, then Idle — this transition in a
// background workspace should trigger a Done sound notification.
@@ -1177,7 +955,12 @@ fn client_receives_notify_on_agent_state_change() {
let mut work_response = String::new();
work_reader.read_line(&mut work_response).unwrap();
- thread::sleep(Duration::from_millis(100));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should stay responsive after working state report"
+ );
let mut idle_stream = UnixStream::connect(&api_socket).expect("connect to API");
let idle_request = format!(
diff --git a/tests/detach_reattach.rs b/tests/detach_reattach.rs
index a6bed866..9eda9e36 100644
--- a/tests/detach_reattach.rs
+++ b/tests/detach_reattach.rs
@@ -4,7 +4,7 @@
mod support;
use std::fs;
-use std::io::{BufRead, BufReader, Read, Write};
+use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::UnixStream;
use std::path::PathBuf;
use std::sync::{Mutex, MutexGuard, OnceLock};
@@ -14,8 +14,9 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize};
use serde_json::Value;
use support::{
- cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid,
- unregister_spawned_herdr_pid,
+ cleanup_test_base, client_handshake, drain_messages, read_server_message, register_runtime_dir,
+ register_spawned_herdr_pid, send_detach, send_input, unregister_spawned_herdr_pid,
+ wait_for_disconnect, wait_for_file, wait_for_message_variant, wait_for_socket, wait_until,
};
fn unique_test_dir() -> PathBuf {
@@ -68,28 +69,6 @@ fn test_lock() -> MutexGuard<'static, ()> {
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
-fn wait_for_socket(path: &PathBuf, timeout: Duration) {
- let deadline = Instant::now() + timeout;
- while Instant::now() < deadline {
- if path.exists() && UnixStream::connect(path).is_ok() {
- return;
- }
- thread::sleep(Duration::from_millis(25));
- }
- panic!("socket did not appear at {}", path.display());
-}
-
-fn wait_for_file(path: &PathBuf, timeout: Duration) {
- let deadline = Instant::now() + timeout;
- while Instant::now() < deadline {
- if path.exists() {
- return;
- }
- thread::sleep(Duration::from_millis(25));
- }
- panic!("file did not appear at {}", path.display());
-}
-
fn spawn_server(
config_home: &PathBuf,
runtime_dir: &PathBuf,
@@ -211,228 +190,6 @@ fn first_pane_id(response: &Value) -> String {
}
// ---------------------------------------------------------------------------
-// Bincode v2 varint encoding helpers
-// ---------------------------------------------------------------------------
-
-/// Encode a varint u32 value according to bincode v2 VarintEncoding.
-fn encode_varint_u32(v: u32) -> Vec<u8> {
- if v < 251 {
- vec![v as u8]
- } else if v < 65536 {
- let mut buf = vec![251u8];
- buf.extend_from_slice(&(v as u16).to_le_bytes());
- buf
- } else {
- let mut buf = vec![252u8];
- buf.extend_from_slice(&v.to_le_bytes());
- buf
- }
-}
-
-/// Encode a varint u16 value.
-fn encode_varint_u16(v: u16) -> Vec<u8> {
- if v < 251 {
- vec![v as u8]
- } else {
- let mut buf = vec![251u8];
- buf.extend_from_slice(&v.to_le_bytes());
- buf
- }
-}
-
-/// Encode an enum variant with its fields.
-fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> {
- let mut buf = encode_varint_u32(variant_idx);
- for field in fields {
- buf.extend_from_slice(field);
- }
- buf
-}
-
-/// Frame a message with u32LE length prefix.
-fn frame_message(payload: &[u8]) -> Vec<u8> {
- let len = payload.len() as u32;
- let mut framed = len.to_le_bytes().to_vec();
- framed.extend_from_slice(payload);
- framed
-}
-
-/// Decode a varint u32 from a byte slice at the given offset.
-fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> {
- if offset >= payload.len() {
- return Err("payload too short for varint".into());
- }
- let first_byte = payload[offset];
- match first_byte {
- 0..=250 => Ok((first_byte as u32, 1)),
- 251 => {
- if offset + 3 > payload.len() {
- return Err("payload too short for u16 varint".into());
- }
- let v = u16::from_le_bytes(
- payload[offset + 1..offset + 3]
- .try_into()
- .map_err(|e: std::array::TryFromSliceError| e.to_string())?,
- );
- Ok((v as u32, 3))
- }
- 252 => {
- if offset + 5 > payload.len() {
- return Err("payload too short for u32 varint".into());
- }
- let v = u32::from_le_bytes(
- payload[offset + 1..offset + 5]
- .try_into()
- .map_err(|e: std::array::TryFromSliceError| e.to_string())?,
- );
- Ok((v, 5))
- }
- _ => Err(format!("unsupported varint tag: {first_byte}")),
- }
-}
-
-/// Sends a Hello message over the client socket and reads the Welcome response.
-fn client_handshake(
- stream: &mut UnixStream,
- version: u32,
- cols: u16,
- rows: u16,
-) -> Result<(u32, Option<String>), String> {
- stream
- .set_read_timeout(Some(Duration::from_secs(5)))
- .map_err(|e| e.to_string())?;
-
- // Encode Hello message: ClientMessage variant 0.
- let hello_payload = encode_varint_enum(
- 0,
- &[
- &encode_varint_u32(version),
- &encode_varint_u16(cols),
- &encode_varint_u16(rows),
- ],
- );
- let framed = frame_message(&hello_payload);
- stream.write_all(&framed).map_err(|e| e.to_string())?;
- stream.flush().map_err(|e| e.to_string())?;
-
- // Read the framed response.
- let mut len_buf = [0u8; 4];
- stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?;
- let len = u32::from_le_bytes(len_buf) as usize;
-
- if len > 2 * 1024 * 1024 {
- return Err(format!("oversized response: {len}"));
- }
-
- let mut payload = vec![0u8; len];
- stream.read_exact(&mut payload).map_err(|e| e.to_string())?;
-
- // Decode Welcome: ServerMessage variant 0 = Welcome { version: u32, error: Option<String> }
- let mut offset = 0;
-
- let (variant, consumed) = decode_varint_u32(&payload, offset)?;
- offset += consumed;
- if variant != 0 {
- return Err(format!(
- "expected Welcome (variant 0), got variant {variant}"
- ));
- }
-
- let (version, consumed) = decode_varint_u32(&payload, offset)?;
- offset += consumed;
-
- if offset >= payload.len() {
- return Err("payload too short for Option tag".into());
- }
- let option_tag = payload[offset];
- offset += 1;
-
- let error = if option_tag == 1 {
- let (str_len, consumed) = decode_varint_u32(&payload, offset)?;
- offset += consumed;
- let str_len = str_len as usize;
-
- if offset + str_len > payload.len() {
- return Err("payload too short for string content".into());
- }
- let s = String::from_utf8(payload[offset..offset + str_len].to_vec())
- .map_err(|e| e.to_string())?;
- Some(s)
- } else {
- None
- };
-
- Ok((version, error))
-}
-
-/// Read a framed ServerMessage from the stream, returning the variant index
-/// and the raw payload for further decoding.
-fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec<u8>), String> {
- stream
- .set_read_timeout(Some(Duration::from_secs(10)))
- .map_err(|e| e.to_string())?;
-
- let mut len_buf = [0u8; 4];
- stream
- .read_exact(&mut len_buf)
- .map_err(|e| format!("read length prefix: {e}"))?;
- let len = u32::from_le_bytes(len_buf) as usize;
-
- if len > 2 * 1024 * 1024 {
- return Err(format!("oversized frame: {len} bytes"));
- }
- if len == 0 {
- return Err("zero-length frame".into());
- }
-
- let mut payload = vec![0u8; len];
- stream
- .read_exact(&mut payload)
- .map_err(|e| format!("read payload: {e}"))?;
-
- let (variant, consumed) = decode_varint_u32(&payload, 0)?;
-
- Ok((variant, payload[consumed..].to_vec()))
-}
-
-/// Sends a ClientMessage::Input with the given raw bytes.
-fn send_input(stream: &mut UnixStream, data: &[u8]) -> Result<(), String> {
- let input_payload = {
- let mut buf = encode_varint_u32(1); // variant 1 = Input
- buf.extend_from_slice(&encode_varint_u32(data.len() as u32));
- buf.extend_from_slice(data);
- buf
- };
- let framed = frame_message(&input_payload);
- stream
- .write_all(&framed)
- .map_err(|e| format!("write input: {e}"))?;
- stream.flush().map_err(|e| format!("flush input: {e}"))?;
- Ok(())
-}
-
-/// Sends a ClientMessage::Detach.
-fn send_detach(stream: &mut UnixStream) -> Result<(), String> {
- // ClientMessage::Detach is variant 3 with no fields.
- let detach_payload = encode_varint_u32(3);
- let framed = frame_message(&detach_payload);
- stream
- .write_all(&framed)
- .map_err(|e| format!("write detach: {e}"))?;
- stream.flush().map_err(|e| format!("flush detach: {e}"))?;
- Ok(())
-}
-
-/// Drains all pending messages from the server stream (non-blocking).
-fn drain_messages(stream: &mut UnixStream) {
- stream
- .set_read_timeout(Some(Duration::from_millis(200)))
- .unwrap();
- while let Ok(_) = read_server_message(stream) {}
- stream.set_read_timeout(None).unwrap();
-}
-
-// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
@@ -468,15 +225,18 @@ fn navigate_q_detaches_client_and_server_persists() {
// Send prefix key (Ctrl+B = 0x02) then 'q' (quit/detach in persistence mode).
send_input(&mut stream, &[0x02]).expect("send prefix");
- thread::sleep(Duration::from_millis(300));
// Drain any frames generated by entering navigate mode.
drain_messages(&mut stream);
send_input(&mut stream, b"q").expect("send detach key");
- // Give the server time to process the detach.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should still respond to ping after client detach"
+ );
// Verify server is still alive and responsive.
let response = ping_socket(&api_socket);
@@ -488,29 +248,9 @@ fn navigate_q_detaches_client_and_server_persists() {
// The client should receive a ServerShutdown with reason "detached"
// shortly after the quit/detach key. There may be some frames in
// between from the mode change, so we read multiple messages.
- stream
- .set_read_timeout(Some(Duration::from_secs(5)))
- .unwrap();
- let mut got_shutdown = false;
- let deadline = Instant::now() + Duration::from_secs(5);
- while Instant::now() < deadline {
- match read_server_message(&mut stream) {
- Ok((variant, _payload)) => {
- if variant == 2 {
- // ServerMessage::ServerShutdown — detach acknowledged.
- got_shutdown = true;
- break;
- }
- // Variant 1 = Frame — may arrive before the shutdown.
- // Keep reading.
- }
- Err(_) => {
- // Connection closed — also acceptable.
- got_shutdown = true; // consider this a successful detach
- break;
- }
- }
- }
+ let got_shutdown = wait_for_message_variant(&mut stream, Duration::from_secs(2), 2)
+ .expect("wait for shutdown message")
+ || wait_for_disconnect(&mut stream, Duration::from_secs(1)).expect("wait for disconnect");
assert!(
got_shutdown,
"client should receive ServerShutdown after quit/detach key"
@@ -548,8 +288,12 @@ fn explicit_detach_message_causes_clean_disconnect() {
// Send ClientMessage::Detach directly.
send_detach(&mut stream).expect("send detach message");
- // Give the server time to process.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should persist after client Detach message"
+ );
// Verify server is still alive.
let response = ping_socket(&api_socket);
@@ -561,25 +305,8 @@ fn explicit_detach_message_causes_clean_disconnect() {
// The client connection should eventually be closed.
// After sending Detach, the server removes the client.
// We may still receive a few queued frames before the connection closes.
- stream
- .set_read_timeout(Some(Duration::from_secs(3)))
- .unwrap();
-
- // Drain remaining messages until EOF or error.
- let mut got_eof = false;
- let deadline = Instant::now() + Duration::from_secs(3);
- while Instant::now() < deadline {
- match read_server_message(&mut stream) {
- Ok((variant, _)) => {
- // May receive frames or other messages before disconnect.
- let _ = variant;
- }
- Err(_) => {
- got_eof = true;
- break;
- }
- }
- }
+ let got_eof =
+ wait_for_disconnect(&mut stream, Duration::from_secs(2)).expect("wait for disconnect");
assert!(
got_eof,
"client connection should be closed after explicit Detach message"
@@ -629,14 +356,15 @@ fn reattach_after_detach_shows_current_state() {
"workspace creation should succeed: {ws_response}"
);
- // Give the server time to process and render.
- thread::sleep(Duration::from_millis(300));
-
// Client A detaches (send ClientMessage::Detach).
send_detach(&mut stream_a).expect("send detach");
- // Give the server time to process the detach.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should persist after detach"
+ );
// Verify server is still alive.
let response = ping_socket(&api_socket);
@@ -739,15 +467,18 @@ fn processes_survive_during_and_after_detach() {
// pane with a shell running.
send_input(&mut stream, b"echo SURVIVED_DETACH\n").expect("send echo command");
- // Wait for the shell to process the command.
- thread::sleep(Duration::from_millis(500));
-
// Drain any frames generated by the input.
drain_messages(&mut stream);
// Detach the client via explicit Detach message.
send_detach(&mut stream).expect("send detach");
- thread::sleep(Duration::from_millis(500));
+
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should persist after detach"
+ );
// Verify server is still alive after detach.
let response = ping_socket(&api_socket);
@@ -756,8 +487,10 @@ fn processes_survive_during_and_after_detach() {
"server should persist after detach: {response}"
);
- // Wait a moment while detached.
- thread::sleep(Duration::from_secs(1));
+ assert!(
+ wait_for_disconnect(&mut stream, Duration::from_secs(2)).expect("wait for detach"),
+ "detached client connection should close"
+ );
// Reattach — verify we can connect and receive a frame.
let mut stream_b = UnixStream::connect(&client_socket).expect("should reattach");
@@ -821,8 +554,12 @@ fn server_persists_after_client_connection_drop() {
// Drop the connection abruptly (simulating client crash).
drop(stream);
- // Give the server time to detect the disconnection.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should persist after client connection drop"
+ );
// Verify server is still alive.
let response = ping_socket(&api_socket);
@@ -873,7 +610,13 @@ fn output_accumulated_while_detached_visible_on_reattach() {
// Detach client A immediately.
send_detach(&mut stream_a).expect("send detach");
- thread::sleep(Duration::from_millis(500));
+
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ ping_socket(&api_socket).contains("pong")
+ }),
+ "server should persist"
+ );
// Verify server alive.
let response = ping_socket(&api_socket);
@@ -887,8 +630,6 @@ fn output_accumulated_while_detached_visible_on_reattach() {
let ws_create_response = workspace_create(&api_socket, "scrollback-test");
assert_eq!(ws_create_response["result"]["type"], "workspace_created");
- thread::sleep(Duration::from_millis(300));
-
// Find the workspace and pane IDs.
let ws_response = workspace_list(&api_socket);
let ws_id = workspace_id_by_label(&ws_response, "scrollback-test");
@@ -907,8 +648,16 @@ fn output_accumulated_while_detached_visible_on_reattach() {
let mut send_response = String::new();
send_reader.read_line(&mut send_response).unwrap();
- // Wait for the echo command to execute.
- thread::sleep(Duration::from_millis(500));
+ assert!(
+ wait_until(Duration::from_secs(2), Duration::from_millis(25), || {
+ let read_response = pane_read_recent(&api_socket, &pane_id);
+ read_response["result"]["read"]["text"]
+ .as_str()
+ .unwrap_or_default()
+ .contains("DURING_DETACH")
+ }),
+ "pane should contain output produced while detached before reattach"
+ );
// --- Client B (reattach) ---
let mut stream_b = UnixStream::connect(&client_socket).expect("client B should connect");
diff --git a/tests/support/mod.rs b/tests/support/mod.rs
index 52a05578..44c25188 100644
--- a/tests/support/mod.rs
+++ b/tests/support/mod.rs
@@ -1,5 +1,9 @@
+#![allow(dead_code)]
+
use std::collections::HashSet;
use std::fs;
+use std::io::{Read, Write};
+use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, Once, OnceLock};
use std::thread;
@@ -66,6 +70,269 @@ pub fn cleanup_test_base(base: &Path) {
let _ = fs::remove_dir_all(base);
}
+pub fn wait_for_socket(path: &Path, timeout: Duration) {
+ let deadline = Instant::now() + timeout;
+ while Instant::now() < deadline {
+ if path.exists() && UnixStream::connect(path).is_ok() {
+ return;
+ }
+ thread::sleep(Duration::from_millis(25));
+ }
+ panic!("socket did not appear at {}", path.display());
+}
+
+pub fn wait_for_file(path: &Path, timeout: Duration) {
+ let deadline = Instant::now() + timeout;
+ while Instant::now() < deadline {
+ if path.exists() {
+ return;
+ }
+ thread::sleep(Duration::from_millis(25));
+ }
+ panic!("file did not appear at {}", path.display());
+}
+
+pub fn encode_varint_u32(v: u32) -> Vec<u8> {
+ if v < 251 {
+ vec![v as u8]
+ } else if v < 65536 {
+ let mut buf = vec![251u8];
+ buf.extend_from_slice(&(v as u16).to_le_bytes());
+ buf
+ } else {
+ let mut buf = vec![252u8];
+ buf.extend_from_slice(&v.to_le_bytes());
+ buf
+ }
+}
+
+pub fn encode_varint_u16(v: u16) -> Vec<u8> {
+ if v < 251 {
+ vec![v as u8]
+ } else {
+ let mut buf = vec![251u8];
+ buf.extend_from_slice(&v.to_le_bytes());
+ buf
+ }
+}
+
+pub fn frame_message(payload: &[u8]) -> Vec<u8> {
+ let len = payload.len() as u32;
+ let mut framed = len.to_le_bytes().to_vec();
+ framed.extend_from_slice(payload);
+ framed
+}
+
+pub fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> {
+ if offset >= payload.len() {
+ return Err("payload too short for varint".into());
+ }
+ let first_byte = payload[offset];
+ match first_byte {
+ 0..=250 => Ok((first_byte as u32, 1)),
+ 251 => {
+ if offset + 3 > payload.len() {
+ return Err("payload too short for u16 varint".into());
+ }
+ let v = u16::from_le_bytes(
+ payload[offset + 1..offset + 3]
+ .try_into()
+ .map_err(|e: std::array::TryFromSliceError| e.to_string())?,
+ );
+ Ok((v as u32, 3))
+ }
+ 252 => {
+ if offset + 5 > payload.len() {
+ return Err("payload too short for u32 varint".into());
+ }
+ let v = u32::from_le_bytes(
+ payload[offset + 1..offset + 5]
+ .try_into()
+ .map_err(|e: std::array::TryFromSliceError| e.to_string())?,
+ );
+ Ok((v, 5))
+ }
+ _ => Err(format!("unsupported varint tag: {first_byte}")),
+ }
+}
+
+fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> {
+ let mut buf = encode_varint_u32(variant_idx);
+ for field in fields {
+ buf.extend_from_slice(field);
+ }
+ buf
+}
+
+fn decode_welcome(payload: &[u8]) -> Result<(u32, Option<String>), String> {
+ let mut offset = 0;
+ let (variant, consumed) = decode_varint_u32(payload, offset)?;
+ offset += consumed;
+ if variant != 0 {
+ return Err(format!(
+ "expected Welcome (variant 0), got variant {variant}"
+ ));
+ }
+
+ let (version, consumed) = decode_varint_u32(payload, offset)?;
+ offset += consumed;
+
+ if offset >= payload.len() {
+ return Err("payload too short for Option tag".into());
+ }
+ let option_tag = payload[offset];
+ offset += 1;
+
+ let error = if option_tag == 1 {
+ let (str_len, consumed) = decode_varint_u32(payload, offset)?;
+ offset += consumed;
+ let str_len = str_len as usize;
+ if offset + str_len > payload.len() {
+ return Err("payload too short for string content".into());
+ }
+ Some(
+ String::from_utf8(payload[offset..offset + str_len].to_vec())
+ .map_err(|e| e.to_string())?,
+ )
+ } else {
+ None
+ };
+
+ Ok((version, error))
+}
+
+pub fn client_handshake(
+ stream: &mut UnixStream,
+ version: u32,
+ cols: u16,
+ rows: u16,
+) -> Result<(u32, Option<String>), String> {
+ stream
+ .set_read_timeout(Some(Duration::from_secs(5)))
+ .map_err(|e| e.to_string())?;
+
+ let hello_payload = encode_varint_enum(
+ 0,
+ &[
+ &encode_varint_u32(version),
+ &encode_varint_u16(cols),
+ &encode_varint_u16(rows),
+ ],
+ );
+ let framed = frame_message(&hello_payload);
+ stream.write_all(&framed).map_err(|e| e.to_string())?;
+ stream.flush().map_err(|e| e.to_string())?;
+
+ let mut len_buf = [0u8; 4];
+ stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?;
+ let len = u32::from_le_bytes(len_buf) as usize;
+ if len > 2 * 1024 * 1024 {
+ return Err(format!("oversized response: {len}"));
+ }
+
+ let mut payload = vec![0u8; len];
+ stream.read_exact(&mut payload).map_err(|e| e.to_string())?;
+ decode_welcome(&payload)
+}
+
+pub fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec<u8>), String> {
+ let mut len_buf = [0u8; 4];
+ stream
+ .read_exact(&mut len_buf)
+ .map_err(|e| format!("read length prefix: {e}"))?;
+ let len = u32::from_le_bytes(len_buf) as usize;
+ if len > 2 * 1024 * 1024 {
+ return Err(format!("oversized frame: {len} bytes"));
+ }
+ if len == 0 {
+ return Err("zero-length frame".into());
+ }
+
+ let mut payload = vec![0u8; len];
+ stream
+ .read_exact(&mut payload)
+ .map_err(|e| format!("read payload: {e}"))?;
+
+ let (variant, consumed) = decode_varint_u32(&payload, 0)?;
+ Ok((variant, payload[consumed..].to_vec()))
+}
+
+pub fn send_input(stream: &mut UnixStream, data: &[u8]) -> Result<(), String> {
+ let mut buf = encode_varint_u32(1);
+ buf.extend_from_slice(&encode_varint_u32(data.len() as u32));
+ buf.extend_from_slice(data);
+ let framed = frame_message(&buf);
+ stream
+ .write_all(&framed)
+ .map_err(|e| format!("write input: {e}"))?;
+ stream.flush().map_err(|e| format!("flush input: {e}"))?;
+ Ok(())
+}
+
+pub fn send_detach(stream: &mut UnixStream) -> Result<(), String> {
+ let detach_payload = encode_varint_u32(3);
+ let framed = frame_message(&detach_payload);
+ stream
+ .write_all(&framed)
+ .map_err(|e| format!("write detach: {e}"))?;
+ stream.flush().map_err(|e| format!("flush detach: {e}"))?;
+ Ok(())
+}
+
+pub fn drain_messages(stream: &mut UnixStream) {
+ stream
+ .set_read_timeout(Some(Duration::from_millis(200)))
+ .unwrap();
+ while let Ok(_) = read_server_message(stream) {}
+ stream.set_read_timeout(None).unwrap();
+}
+
+pub fn wait_until<F>(timeout: Duration, interval: Duration, mut predicate: F) -> bool
+where
+ F: FnMut() -> bool,
+{
+ let deadline = Instant::now() + timeout;
+ while Instant::now() < deadline {
+ if predicate() {
+ return true;
+ }
+ thread::sleep(interval);
+ }
+ predicate()
+}
+
+pub fn wait_for_message_variant(
+ stream: &mut UnixStream,
+ timeout: Duration,
+ variant: u32,
+) -> Result<bool, String> {
+ stream
+ .set_read_timeout(Some(Duration::from_millis(200)))
+ .map_err(|e| e.to_string())?;
+ let deadline = Instant::now() + timeout;
+ while Instant::now() < deadline {
+ match read_server_message(stream) {
+ Ok((got, _)) if got == variant => return Ok(true),
+ Ok(_) => continue,
+ Err(_) => continue,
+ }
+ }
+ Ok(false)
+}
+
+pub fn wait_for_disconnect(stream: &mut UnixStream, timeout: Duration) -> Result<bool, String> {
+ stream
+ .set_read_timeout(Some(Duration::from_millis(200)))
+ .map_err(|e| e.to_string())?;
+ let deadline = Instant::now() + timeout;
+ while Instant::now() < deadline {
+ if read_server_message(stream).is_err() {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+}
+
pub fn cleanup_registered_herdr_pids() {
let pids: Vec<u32> = {
let mut registry = pid_registry_lock();
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-06 22:27:56 +0000