diff --git a/docs/plans/project-setup.md b/docs/plans/project-setup.md index 839f07b..5589a70 100644 --- a/docs/plans/project-setup.md +++ b/docs/plans/project-setup.md @@ -2,7 +2,7 @@ Status: draft (revised 2026-08-17 to reflect landed upstream changes; Phase 4 landed 2026-08-17 — architecture docs + BAST schema + -renumbered ADRs) +renumbered ADRs; Phase 5 landed 2026-08-17 — tests) Last updated: 2026-08-17 ## Overview @@ -488,18 +488,47 @@ split (`STREAM_CTRL_IN` = 3, `STREAM_CTRL_OUT` = 4) is an amendment inside ADR-001, mirroring alknet (it was not a standalone ADR there either). `AGENTS.md` was updated to match this mapping. -### Phase 5: Tests +### Phase 5: Tests — landed 2026-08-17 -1. Port existing unit tests from `alknet-tty` (wire, negotiation, - control, adapter) — these live inline in each `src/*.rs` module's - `#[cfg(test)] mod tests`. -2. Port integration tests from `alknet-tty-local/tests/` (negotiation, - pipe, pty) into `tests/` at the crate root. The pty test stays - `#[cfg(unix)]`. -3. Add channels integration tests — end-to-end `register_openable` + - `ChannelClient::open_channel` + `drive_session` round-trip. Use - the in-memory `MockBackend` from `backend.rs` for the producer - side so the test doesn't need a real PTY. +1. Ported existing unit tests from `alknet-tty` (wire, negotiation, + control, adapter) inline in each `src/*.rs` module's + `#[cfg(test)] mod tests` — done in Phases 1–2 alongside the + production code (80 lib tests, all passing). +2. Ported integration tests from `alknet-tty-local/tests/` into + `tests/` at the crate root: + - [`tests/common/mod.rs`](../tests/common/mod.rs) — the + `ClientSide` wire-protocol harness + `spawn_session` helper + + `negotiate_pty_json` / `negotiate_pipe_json` builders. Imports + renamed `alknet_core::auth::Identity` → + `alkcall::core::auth::Identity`, `alknet_tty::...` → + `alktty::...`. + - [`tests/negotiation.rs`](../tests/negotiation.rs) — 4 + negotiation-error scenarios (unknown_backend, + malformed_negotiation ×3, allocate_failed). + - [`tests/pipe.rs`](../tests/pipe.rs) — 6 pipe-mode scenarios + (echo happy path, separate stderr, SIGTERM, cancel cleanup, + resize no-op, stdout sentinel). The 2 cancel-cleanup / + SIGTERM tests are `#[cfg(unix)]`. + - [`tests/pty.rs`](../tests/pty.rs) — 8 PTY-mode scenarios + (echo, interactive cat, resize, SIGINT, process-group + signal, stdin-EOF sentinel, cancel cleanup, + exit-chunk-is-last). The 4 signal / cancel-cleanup tests + are `#[cfg(unix)]`. + Each test file carries `#![cfg(feature = "local")]` so the + default crate (no features) skips the integration binaries + and stays wasm-buildable; `cargo test --all-features` runs + all 19 integration tests (5 + 6 + 8). +3. Added channels integration tests inline in + [`src/channels.rs`](../src/channels.rs) `mod tests` (done in + Phase 2 alongside the producer code): `register_openable` + registration, end-to-end `ChannelClient::call_open_op` returns + `channel_id`, scope-gate denies without `tty:open`, and the + `MockBackend` exit-code sanity check. Uses the in-memory + `MockBackend` from `backend.rs` for the producer side so no + real PTY is needed. + +Total: 80 lib tests + 19 integration tests = 99 passing under +`--all-features`; 80 lib tests under default (wasm-clean) build. ## Open Questions diff --git a/tests/common/mod.rs b/tests/common/mod.rs new file mode 100644 index 0000000..57e6817 --- /dev/null +++ b/tests/common/mod.rs @@ -0,0 +1,291 @@ +//! Test harness: client-side wire protocol helpers + `drive_session` +//! stand-in over a `tokio::io::duplex` pair. +//! +//! Mirrors the `MockBackend` test harness in `src/adapter.rs`, but +//! drives a real `LocalTtyBackend` instead of a mock. The test acts as +//! the client: writes the negotiation frame, sends stdin/control +//! chunks, reads stdout/stderr/control chunks, asserts the exit chunk +//! and stream close (ADR-052, ADR-055, ADR-056). +//! +//! `allow(dead_code)` is needed because each test binary uses a +//! different subset of the helpers; clippy would otherwise flag the +//! unused ones (the helpers are part of the shared client-side wire +//! protocol toolkit the task requires). + +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Arc; + +use alkcall::core::auth::Identity; +use alktty::adapter::drive_session; +use alktty::backend::TtyBackend; +use alktty::wire::{ChunkReader, STREAM_CTRL_IN, STREAM_CTRL_OUT, STREAM_STDERR, STREAM_STDOUT}; +use bytes::Bytes; +use tokio::io::duplex; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +/// The scope the test identity carries (matches `TtyAdapter::TTY_OPEN_SCOPE`). +pub const TTY_OPEN_SCOPE: &str = "tty:open"; + +/// A client-side negotiator: writes the JSON negotiation frame and +/// provides helpers for sending stdin chunks and control messages, +/// and for reading chunks/error frames the server writes back. +pub struct ClientSide { + pub write: tokio::io::WriteHalf, + pub read: tokio::io::ReadHalf, +} + +impl ClientSide { + /// Write a length-prefixed JSON negotiation frame (the Phase 1 + /// carriage). The `body` should already be a complete JSON + /// `NegotiateRequest` object. + pub async fn write_negotiation(&mut self, body: &str) { + let len = body.len() as u32; + self.write.write_all(&len.to_be_bytes()).await.unwrap(); + self.write.write_all(body.as_bytes()).await.unwrap(); + self.write.flush().await.unwrap(); + } + + /// Write a length-prefixed raw bytes negotiation frame (for + /// malformed-JSON tests). + pub async fn write_negotiation_bytes(&mut self, body: &[u8]) { + let len = body.len() as u32; + self.write.write_all(&len.to_be_bytes()).await.unwrap(); + self.write.write_all(body).await.unwrap(); + self.write.flush().await.unwrap(); + } + + /// Write a raw chunk: `[stream_type: u8][len: u32 be][payload]`. + pub async fn write_chunk(&mut self, stream_type: u8, payload: &[u8]) { + let mut header = [0u8; 5]; + header[0] = stream_type; + let len = payload.len() as u32; + header[1..].copy_from_slice(&len.to_be_bytes()); + self.write.write_all(&header).await.unwrap(); + if !payload.is_empty() { + self.write.write_all(payload).await.unwrap(); + } + self.write.flush().await.unwrap(); + } + + /// Write a client→server control chunk (`STREAM_CTRL_IN`, stream_type + /// 3) carrying a serialized `ControlMessage` JSON payload (`Resize`, + /// `Signal`, or `Eof`). + pub async fn write_control(&mut self, json: &[u8]) { + self.write_chunk(STREAM_CTRL_IN, json).await; + } + + /// Read one raw chunk from the server. Returns the `stream_type` + /// and the payload bytes, or `None` when the server closed the + /// stream cleanly (EOF). + pub async fn try_read_chunk(&mut self) -> Option<(u8, Bytes)> { + let mut reader = ChunkReader::new(&mut self.read); + match reader.read_chunk().await { + Ok(chunk) => Some((chunk.stream_type, chunk.bytes)), + Err(_) => None, + } + } + + /// Read one chunk, panicking on read errors (used when the test + /// expects a chunk, not a close). + pub async fn read_chunk(&mut self) -> (u8, Bytes) { + self.try_read_chunk() + .await + .expect("expected a chunk, got stream close") + } + + /// Read one chunk with a timeout. Returns `None` on timeout or + /// stream close. + pub async fn read_chunk_timeout( + &mut self, + timeout: std::time::Duration, + ) -> Option<(u8, Bytes)> { + tokio::time::timeout(timeout, self.try_read_chunk()) + .await + .unwrap_or_default() + } + + /// Read a length-prefixed JSON error frame (Phase 1 framing, used + /// for negotiation errors). The first byte of the 4-byte length + /// prefix MUST be `0x00` (the framing-disambiguation invariant — + /// ADR-052 §5); this method asserts that. Returns the parsed JSON. + pub async fn read_error_frame(&mut self) -> serde_json::Value { + let mut first = [0u8; 1]; + self.read.read_exact(&mut first).await.unwrap(); + assert_eq!( + first[0], 0x00, + "error frame length prefix high byte must be 0x00 (ADR-052 §5)" + ); + let mut len_rest = [0u8; 3]; + self.read.read_exact(&mut len_rest).await.unwrap(); + let len = u32::from_be_bytes([first[0], len_rest[0], len_rest[1], len_rest[2]]) as usize; + let mut body = vec![0u8; len]; + self.read.read_exact(&mut body).await.unwrap(); + serde_json::from_slice(&body).unwrap() + } + + /// Read chunks until the exit control chunk is observed, then + /// return the accumulated stdout bytes, stderr bytes, and the + /// exit code. Returns `None` if the stream closes before an exit + /// chunk arrives. + pub async fn read_until_exit(&mut self) -> Option<(Vec, Vec, i32)> { + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + loop { + let (st, bytes) = self.try_read_chunk().await?; + match st { + STREAM_STDOUT => { + if !bytes.is_empty() { + stdout.extend_from_slice(&bytes); + } + } + STREAM_STDERR => { + if !bytes.is_empty() { + stderr.extend_from_slice(&bytes); + } + } + STREAM_CTRL_OUT => { + let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + if v["type"] == "exit" { + return Some((stdout, stderr, v["code"].as_i64().unwrap() as i32)); + } + } + other => panic!("unexpected stream_type {other} from server"), + } + } + } + + /// Same as [`read_until_exit`](Self::read_until_exit) but with a + /// timeout on each individual chunk read. + pub async fn read_until_exit_timeout( + &mut self, + per_chunk_timeout: std::time::Duration, + ) -> Option<(Vec, Vec, i32)> { + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + loop { + let (st, bytes) = self.read_chunk_timeout(per_chunk_timeout).await?; + match st { + STREAM_STDOUT => { + if !bytes.is_empty() { + stdout.extend_from_slice(&bytes); + } + } + STREAM_STDERR => { + if !bytes.is_empty() { + stderr.extend_from_slice(&bytes); + } + } + STREAM_CTRL_OUT => { + let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + if v["type"] == "exit" { + return Some((stdout, stderr, v["code"].as_i64().unwrap() as i32)); + } + } + other => panic!("unexpected stream_type {other} from server"), + } + } + } + + /// Read chunks until the exit chunk is observed, then assert + /// the stream closes cleanly and NO further chunks arrive + /// (ADR-055 exit-chunk-is-last invariant). Returns the stdout, + /// stderr, and exit code. + pub async fn read_until_exit_and_close(&mut self) -> (Vec, Vec, i32) { + let (stdout, stderr, code) = self + .read_until_exit() + .await + .expect("expected exit chunk before stream close"); + self.assert_no_more_chunks().await; + (stdout, stderr, code) + } + + /// After the exit chunk, assert the server closes the stream + /// without sending any further chunks. Panics if a chunk arrives. + pub async fn assert_no_more_chunks(&mut self) { + match self + .read_chunk_timeout(std::time::Duration::from_millis(500)) + .await + { + None => {} + Some((st, bytes)) => panic!( + "chunk arrived after exit chunk: stream_type={st}, bytes={bytes:?} (ADR-055)" + ), + } + } + + /// Close the client write half cleanly. The server's read half sees + /// a clean EOF on its `ChunkReader`, which the adapter treats as a + /// client-cancel of the input direction (it signals EOF to the + /// backend's stdin and stops the input pump). For pipe-mode + /// backends, this is the path that actually closes the child's + /// stdin pipe (tokio's `ChildStdin::poll_shutdown` is a no-op on + /// Unix; dropping the `ChildStdin` is what closes the pipe, and + /// the input pump drops it on ConnectionClosed). + pub async fn close_write_half(&mut self) { + let _ = self.write.shutdown().await; + } +} + +/// Build a `ClientSide` + spawned `drive_session` task over a +/// `tokio::io::duplex` pair. The single `backend` is registered under +/// the given key. The test identity carries `tty:open`, so the +/// adapter's scope-gate passes. +pub fn spawn_session( + backend_key: &str, + backend: Arc, +) -> (ClientSide, tokio::task::JoinHandle<()>) { + let mut backends: HashMap> = HashMap::new(); + backends.insert(backend_key.to_string(), backend); + spawn_session_with_backends(backends) +} + +/// Like [`spawn_session`] but accepts the full backend map (for +/// `unknown_backend` tests). +pub fn spawn_session_with_backends( + backends: HashMap>, +) -> (ClientSide, tokio::task::JoinHandle<()>) { + let (client, server) = duplex(8 * 1024); + let (client_read, client_write) = tokio::io::split(client); + let client_side = ClientSide { + write: client_write, + read: client_read, + }; + let backends = Arc::new(backends); + let identity = Some(test_identity()); + let handle = tokio::spawn(async move { + let (server_read, server_write) = tokio::io::split(server); + drive_session(server_write, server_read, backends, None, identity).await; + }); + (client_side, handle) +} + +/// A test identity carrying the `tty:open` scope, so the adapter's +/// scope-gate at negotiation passes. +pub fn test_identity() -> Identity { + Identity { + id: "test-user".to_string(), + scopes: vec![TTY_OPEN_SCOPE.to_string()], + resources: HashMap::new(), + } +} + +/// Build a negotiation frame JSON string for a PTY-mode session. +pub fn negotiate_pty_json(backend: &str, cmd: &[&str]) -> String { + let cmd_json: Vec = cmd.iter().map(|c| format!("\"{}\"", c)).collect(); + format!( + r#"{{"carriage":"raw","backend":"{backend}","tty":{{"cols":80,"rows":24,"pixel_width":0,"pixel_height":0}},"cmd":[{cmd}]}}"#, + cmd = cmd_json.join(",") + ) +} + +/// Build a negotiation frame JSON string for a pipe-mode session +/// (`tty: null`). +pub fn negotiate_pipe_json(backend: &str, cmd: &[&str]) -> String { + let cmd_json: Vec = cmd.iter().map(|c| format!("\"{}\"", c)).collect(); + format!( + r#"{{"carriage":"raw","backend":"{backend}","tty":null,"cmd":[{cmd}]}}"#, + cmd = cmd_json.join(",") + ) +} diff --git a/tests/negotiation.rs b/tests/negotiation.rs new file mode 100644 index 0000000..95c02f2 --- /dev/null +++ b/tests/negotiation.rs @@ -0,0 +1,116 @@ +#![cfg(feature = "local")] + +//! Negotiation error tests — scenarios 14–18 of +//! `tasks/tty/integration-test.md`. +//! +//! 14. unknown_backend: error response, first byte `0x00` +//! 15. malformed_negotiation (bad JSON): error response +//! 16. malformed_negotiation (carriage != raw): error response +//! 17. malformed_negotiation (empty cmd): error response +//! 18. allocate_failed (nonexistent binary): error response +//! +//! The test identity carries the `tty:open` scope, so the +//! `unknown_backend` test reaches the backend lookup; the scope-gate +//! is exercised in the adapter's unit tests. These tests assert the +//! server sends a length-prefixed JSON error frame (Phase 1 framing, +//! ADR-052) and closes the stream without entering raw mode. + +mod common; + +use std::sync::Arc; + +use alktty::local::LocalTtyBackend; +use common::{negotiate_pipe_json, spawn_session}; + +/// 14. unknown_backend: negotiate `{backend:"kubernetes",...}`, assert +/// the error response `{"error":"unknown_backend","backend":"kubernetes"}`, +/// the first byte of the error frame is `0x00`, and the stream closes. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn unknown_backend_error() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation( + r#"{"carriage":"raw","backend":"kubernetes","tty":null,"cmd":["echo","hi"]}"#, + ) + .await; + + let err = client.read_error_frame().await; + assert_eq!(err["error"], "unknown_backend"); + assert_eq!(err["backend"], "kubernetes"); + + client.assert_no_more_chunks().await; + let _ = server.await; +} + +/// 15. malformed_negotiation (bad JSON): write garbage bytes as the +/// negotiation frame, assert `malformed_negotiation` error response. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn malformed_negotiation_bad_json() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client.write_negotiation("not valid json").await; + + let err = client.read_error_frame().await; + assert_eq!(err["error"], "malformed_negotiation"); + + client.assert_no_more_chunks().await; + let _ = server.await; +} + +/// 16. malformed_negotiation (carriage != raw): negotiate +/// `{carriage:"json",...}`, assert `malformed_negotiation`. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn malformed_negotiation_carriage_not_raw() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation( + r#"{"carriage":"json","backend":"local","tty":null,"cmd":["echo","hi"]}"#, + ) + .await; + + let err = client.read_error_frame().await; + assert_eq!(err["error"], "malformed_negotiation"); + + client.assert_no_more_chunks().await; + let _ = server.await; +} + +/// 17. malformed_negotiation (empty cmd): negotiate `{cmd:[]}`, assert +/// `malformed_negotiation`. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn malformed_negotiation_empty_cmd() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(r#"{"carriage":"raw","backend":"local","tty":null,"cmd":[]}"#) + .await; + + let err = client.read_error_frame().await; + assert_eq!(err["error"], "malformed_negotiation"); + + client.assert_no_more_chunks().await; + let _ = server.await; +} + +/// 18. allocate_failed (nonexistent binary): negotiate with a +/// nonexistent command path, assert `allocate_failed` (the spawn +/// fails). The adapter sends the error response in negotiation +/// framing and closes. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn allocate_failed_nonexistent_binary() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation( + negotiate_pipe_json("local", &["/nonexistent/binary/that/does/not/exist"]).as_str(), + ) + .await; + + let err = client.read_error_frame().await; + assert_eq!(err["error"], "allocate_failed"); + + client.assert_no_more_chunks().await; + let _ = server.await; +} diff --git a/tests/pipe.rs b/tests/pipe.rs new file mode 100644 index 0000000..0438872 --- /dev/null +++ b/tests/pipe.rs @@ -0,0 +1,224 @@ +#![cfg(feature = "local")] + +//! End-to-end integration tests for pipe mode (`terminal: None`) — +//! `LocalTtyBackend` + `TtyAdapter::drive_session` over a +//! `tokio::io::duplex` transport stand-in, running real commands. +//! +//! Covers scenarios 9–13 of `tasks/tty/integration-test.md`: +//! 9. Happy path (echo): stdout "hello", stderr empty, exit 0 +//! 10. Separate stderr: stdout "out", stderr "err", exit 0 +//! 11. Signal (SIGTERM, Unix): exit signal-terminated +//! 12. Cancel cleanup: drop duplex → child killed +//! 13. Resize no-op: control chunk accepted, no error + +mod common; + +use std::sync::Arc; +use std::time::Duration; + +use alktty::local::LocalTtyBackend; +use alktty::wire::STREAM_STDOUT; +use common::{negotiate_pipe_json, spawn_session}; + +/// 9. Happy path (echo): negotiate `{backend:"local", tty:null, +/// cmd:["echo","hello"]}`, read stdout chunks, assert "hello", exit 0. +/// Assert stderr is empty. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pipe_happy_path_echo() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pipe_json("local", &["echo", "hello"]).as_str()) + .await; + + let (stdout, stderr, code) = client + .read_until_exit() + .await + .expect("expected exit chunk before stream close"); + let out = String::from_utf8_lossy(&stdout); + assert!( + out.contains("hello"), + "stdout should contain 'hello'; got: {out:?}" + ); + assert!(stderr.is_empty(), "stderr should be empty; got: {stderr:?}"); + assert_eq!(code, 0, "echo should exit 0"); + client.assert_no_more_chunks().await; + let _ = server.await; +} + +/// 10. Separate stderr: negotiate `cmd:["sh","-c","echo out; echo err >&2"]`, +/// assert stdout stream receives "out", stderr stream receives "err" +/// (as stderr chunks, stream_type 2), exit 0. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pipe_separate_stderr() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation( + negotiate_pipe_json("local", &["sh", "-c", "echo out; echo err >&2"]).as_str(), + ) + .await; + + let (stdout, stderr, code) = client + .read_until_exit() + .await + .expect("expected exit chunk before stream close"); + let out = String::from_utf8_lossy(&stdout); + let err = String::from_utf8_lossy(&stderr); + assert!( + out.contains("out"), + "stdout should contain 'out'; got: {out:?}" + ); + assert!( + err.contains("err"), + "stderr should contain 'err'; got: {err:?}" + ); + assert_eq!(code, 0, "sh should exit 0"); + let _ = server.await; +} + +/// 11. Signal (SIGTERM, Unix): negotiate `cmd:["sleep","60"]`, send +/// `signal:"TERM"`, await exit, assert signal-terminated. +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pipe_signal_sigterm_kills_child() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pipe_json("local", &["sleep", "60"]).as_str()) + .await; + + tokio::time::sleep(Duration::from_millis(150)).await; + + client + .write_control(br#"{"type":"signal","name":"TERM"}"#) + .await; + + let (_out, _err, code) = client + .read_until_exit_timeout(Duration::from_secs(5)) + .await + .expect("expected exit chunk after SIGTERM"); + assert_ne!( + code, 0, + "child killed by SIGTERM should report non-zero exit; got {code}" + ); + let _ = server.await; +} + +/// 12. Cancel cleanup (ADR-056): negotiate `sleep 60`, drop the duplex +/// mid-session, assert the child is killed (no orphan). The child +/// writes its pid to a temp file so we can probe it after the drop. +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pipe_cancel_cleanup_kills_child_no_orphan() { + let pid_file = std::env::temp_dir().join(format!( + "alktty_pipe_cancel_pid_{}_{}.txt", + std::process::id(), + nanos_seed() + )); + let cmd = format!("echo $$ > '{}'; exec sleep 60", pid_file.display()); + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pipe_json("local", &["sh", "-c", cmd.as_str()]).as_str()) + .await; + + for _ in 0..200 { + if pid_file.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + let pid_str = std::fs::read_to_string(&pid_file).expect("pid file written"); + let pid: i32 = pid_str.trim().parse().expect("pid parses"); + let _ = std::fs::remove_file(&pid_file); + + tokio::time::sleep(Duration::from_millis(150)).await; + + drop(client); + server.abort(); + let _ = server.await; + + let mut alive = true; + for _ in 0..100 { + let r = unsafe { libc::kill(pid, 0) }; + if r != 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) { + alive = false; + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(!alive, "child (pid={pid}) should be killed after cancel"); +} + +/// 13. Resize no-op: negotiate `cmd:["cat"]`, send a `resize` control +/// chunk, assert no error (PipeControl::resize is a no-op). Close the +/// write half to signal end-of-input (the adapter's input pump drops +/// the `ChildStdin` on `ConnectionClosed`, which closes the pipe — +/// tokio's `ChildStdin::poll_shutdown` is a no-op on Unix). Await exit. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pipe_resize_noop() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pipe_json("local", &["cat"]).as_str()) + .await; + + tokio::time::sleep(Duration::from_millis(150)).await; + + client + .write_control(br#"{"type":"resize","cols":120,"rows":40}"#) + .await; + + client.write_control(br#"{"type":"eof"}"#).await; + client.close_write_half().await; + + let (_out, _err, code) = client + .read_until_exit_timeout(Duration::from_secs(5)) + .await + .expect("expected exit chunk after resize + eof"); + assert_eq!( + code, 0, + "cat should exit 0 after no-op resize + write-half close" + ); + + let _ = server.await; +} + +/// Sanity: an `echo` in pipe mode should produce at least one +/// stdout chunk with non-empty bytes (the adapter emits a +/// zero-length stdout sentinel after the backend stream ends). +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pipe_echo_emits_stdout_chunk_then_sentinel() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pipe_json("local", &["echo", "hi"]).as_str()) + .await; + + let mut saw_nonempty_stdout = false; + while let Some((st, bytes)) = client.read_chunk_timeout(Duration::from_secs(5)).await { + if st == STREAM_STDOUT && !bytes.is_empty() { + saw_nonempty_stdout = true; + } + if st == alktty::wire::STREAM_CTRL_OUT { + let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + if v["type"] == "exit" { + break; + } + } + } + assert!( + saw_nonempty_stdout, + "expected at least one non-empty stdout chunk" + ); + let _ = server.await; +} + +fn nanos_seed() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0) +} diff --git a/tests/pty.rs b/tests/pty.rs new file mode 100644 index 0000000..6eecf18 --- /dev/null +++ b/tests/pty.rs @@ -0,0 +1,268 @@ +#![cfg(feature = "local")] + +//! End-to-end integration tests for PTY mode (`terminal: Some`) — +//! `LocalTtyBackend` + `TtyAdapter::drive_session` over a +//! `tokio::io::duplex` transport stand-in, running real commands. +//! +//! Covers scenarios 1–8 of `tasks/tty/integration-test.md`: +//! 1. Happy path (echo): stdout contains "hello", exit 0, exit chunk last +//! 2. Interactive (cat): stdin round-trips, eof → exit 0 +//! 3. Resize: control chunk accepted, no error +//! 4. Signal (SIGINT, Unix): exit code signal-terminated, child reaped +//! 5. Process-group signal (Unix): bash -c "sleep 60", INT reaches sleep child +//! 6. Stdin EOF (zero-length chunk): backend stdin closes, exit chunk sent +//! 7. Cancel cleanup (ADR-056): drop duplex → child killed, no orphan +//! 8. Exit-chunk-is-last: no stdout chunk after exit chunk + +mod common; + +use std::sync::Arc; +use std::time::Duration; + +use alktty::local::LocalTtyBackend; +use alktty::wire::{STREAM_CTRL_OUT, STREAM_STDIN}; +use common::{negotiate_pty_json, spawn_session}; + +const PTY_NEG_ECHO: &str = r#"{"carriage":"raw","backend":"local","tty":{"cols":80,"rows":24,"pixel_width":0,"pixel_height":0},"cmd":["echo","hello"]}"#; + +/// 1. Happy path (echo): stdout contains "hello", exit 0, exit chunk is +/// the last chunk before stream close (ADR-055). +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_happy_path_echo() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client.write_negotiation(PTY_NEG_ECHO).await; + + let (stdout, _stderr, code) = client + .read_until_exit() + .await + .expect("expected exit chunk before stream close"); + let s = String::from_utf8_lossy(&stdout); + assert!( + s.contains("hello"), + "stdout should contain 'hello'; got: {s:?}" + ); + assert_eq!(code, 0, "echo should exit 0"); + client.assert_no_more_chunks().await; + let _ = server.await; +} + +/// 2. Interactive (cat): write stdin chunks, then send `eof` and +/// drain stdout. The adapter's drainer writes chunks to the client +/// after the exit chunk resolves, so the test sends all input first +/// (closing cat's stdin with `eof`), then reads the echoed stdout +/// and the exit chunk together. Asserts the echoed "ping" appears in +/// stdout and cat exits 0. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_interactive_cat_round_trip() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pty_json("local", &["cat"]).as_str()) + .await; + + tokio::time::sleep(Duration::from_millis(200)).await; + + client.write_chunk(STREAM_STDIN, b"ping\n").await; + client.write_control(br#"{"type":"eof"}"#).await; + + let (stdout, _stderr, code) = client + .read_until_exit() + .await + .expect("expected exit chunk before stream close"); + let s = String::from_utf8_lossy(&stdout); + assert!( + s.contains("ping"), + "stdin did not round-trip via the PTY echo; got: {s:?}" + ); + assert_eq!(code, 0, "cat should exit 0 on eof"); + let _ = server.await; +} + +/// 3. Resize: send a `resize` control chunk mid-session; assert no +/// error (the PTY resizes). Send `eof`, await exit. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_resize_no_error() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pty_json("local", &["cat"]).as_str()) + .await; + + tokio::time::sleep(Duration::from_millis(150)).await; + + client + .write_control(br#"{"type":"resize","cols":120,"rows":40}"#) + .await; + + client.write_control(br#"{"type":"eof"}"#).await; + + let (_out, _err, code) = client + .read_until_exit_timeout(Duration::from_secs(5)) + .await + .expect("expected exit chunk"); + assert_eq!(code, 0, "cat should exit 0 after resize + eof"); + let _ = server.await; +} + +/// 4. Signal (SIGINT, Unix): negotiate `sleep 60`, send `signal:"INT"`, +/// await the exit chunk. Assert exit code is signal-terminated (non-zero, +/// negative on Unix). Assert the child is reaped (no zombie). +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_signal_sigint_kills_child() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pty_json("local", &["sleep", "60"]).as_str()) + .await; + + tokio::time::sleep(Duration::from_millis(200)).await; + + client + .write_control(br#"{"type":"signal","name":"INT"}"#) + .await; + + let (_out, _err, code) = client + .read_until_exit_timeout(Duration::from_secs(5)) + .await + .expect("expected exit chunk after signal"); + assert_ne!( + code, 0, + "child killed by SIGINT should report non-zero exit; got {code}" + ); + let _ = server.await; +} + +/// 5. Process-group signal (Unix): negotiate `bash -c "sleep 60"`, +/// send `signal:"INT"`, assert the `sleep` child also receives the +/// signal (the process group is targeted — REQ-TTY-02). +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_signal_reaches_process_group_child() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pty_json("local", &["bash", "-c", "sleep 60"]).as_str()) + .await; + + tokio::time::sleep(Duration::from_millis(250)).await; + + client + .write_control(br#"{"type":"signal","name":"INT"}"#) + .await; + + let (_out, _err, code) = client + .read_until_exit_timeout(Duration::from_secs(5)) + .await + .expect("expected exit chunk after group signal"); + assert_ne!( + code, 0, + "process group should have been killed (REQ-TTY-02); got {code}" + ); + let _ = server.await; +} + +/// 6. Stdin EOF (zero-length chunk): negotiate `cat`, send a +/// zero-length stdin chunk (the sentinel), assert the backend's stdin +/// closes, stdout drains, exit chunk is sent. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_stdin_eof_zero_length_chunk() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pty_json("local", &["cat"]).as_str()) + .await; + + tokio::time::sleep(Duration::from_millis(150)).await; + + client.write_chunk(STREAM_STDIN, b"").await; + + let (_out, _err, code) = client + .read_until_exit_timeout(Duration::from_secs(5)) + .await + .expect("expected exit chunk after zero-length stdin sentinel"); + assert_eq!(code, 0, "cat should exit 0 on stdin EOF"); + let _ = server.await; +} + +/// 7. Cancel cleanup (ADR-056): negotiate `sleep 60`, drop the duplex +/// (simulating connection drop) mid-session, assert the child is +/// killed (no orphan). The child writes its pid to a temp file so we +/// can probe it after the drop. +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_cancel_cleanup_kills_child_no_orphan() { + let pid_file = std::env::temp_dir().join(format!( + "alktty_pty_cancel_pid_{}_{}.txt", + std::process::id(), + nanos_seed() + )); + let cmd = format!("echo $$ > '{}'; exec sleep 60", pid_file.display()); + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client + .write_negotiation(negotiate_pty_json("local", &["bash", "-c", cmd.as_str()]).as_str()) + .await; + + for _ in 0..200 { + if pid_file.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + let pid_str = std::fs::read_to_string(&pid_file).expect("pid file written"); + let pid: i32 = pid_str.trim().parse().expect("pid parses"); + let _ = std::fs::remove_file(&pid_file); + + tokio::time::sleep(Duration::from_millis(150)).await; + + drop(client); + server.abort(); + let _ = server.await; + + let mut alive = true; + for _ in 0..100 { + let r = unsafe { libc::kill(pid, 0) }; + if r != 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) { + alive = false; + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(!alive, "child (pid={pid}) should be killed after cancel"); +} + +/// 8. Exit-chunk-is-last: in the happy path, assert no stdout chunk +/// arrives after the exit chunk. Read all chunks, find the exit chunk, +/// assert it is the last chunk before stream close. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pty_exit_chunk_is_last() { + let backend = Arc::new(LocalTtyBackend::new()); + let (mut client, server) = spawn_session("local", backend); + client.write_negotiation(PTY_NEG_ECHO).await; + + let mut saw_exit = false; + while let Some((st, bytes)) = client.read_chunk_timeout(Duration::from_secs(5)).await { + if saw_exit { + panic!("chunk arrived after exit: stream_type={st}, bytes={bytes:?} (ADR-055)"); + } + if st == STREAM_CTRL_OUT { + let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + if v["type"] == "exit" { + assert_eq!(v["code"], 0); + saw_exit = true; + } + } + } + assert!(saw_exit, "did not see the exit chunk"); + let _ = server.await; +} + +fn nanos_seed() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0) +}