diff --git a/tests/e2e-cucumber/src/lib.rs b/tests/e2e-cucumber/src/lib.rs index a5a71d50..b08a7cd3 100644 --- a/tests/e2e-cucumber/src/lib.rs +++ b/tests/e2e-cucumber/src/lib.rs @@ -10,6 +10,8 @@ pub mod loopback_http; pub mod mock_server; pub mod model_id; pub mod panic_capture; +pub mod reader_failure; +pub mod send_until; pub mod serve_log; /// Render everything known about a failed `rocm` invocation. diff --git a/tests/e2e-cucumber/src/reader_failure.rs b/tests/e2e-cucumber/src/reader_failure.rs new file mode 100644 index 00000000..cfe04c4f --- /dev/null +++ b/tests/e2e-cucumber/src/reader_failure.rs @@ -0,0 +1,103 @@ +// Copyright © Advanced Micro Devices, Inc., or its affiliates. +// +// SPDX-License-Identifier: MIT + +//! Failure state shared between a background terminal reader and its waiters. + +use std::sync::Mutex; + +#[derive(Debug, Default)] +struct State { + failed: bool, + message: Option, +} + +/// Atomic observation of the reader's failure state. +#[derive(Debug, PartialEq, Eq)] +pub enum ReaderFailureObservation { + /// The reader has not published a failure. + Running, + /// A failure was published, but its diagnostic was already consumed. + FailedWithoutMessage, + /// A newly observed failure diagnostic. + Message(String), +} + +/// Stores persistent reader-failure state and a diagnostic for one-time reporting. +#[derive(Debug, Default)] +pub struct ReaderFailure { + state: Mutex, +} + +impl ReaderFailure { + /// Publish a reader failure for a waiter to report. + pub fn publish(&self, message: String) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.failed = true; + state.message = Some(message); + } + + /// Take the failure diagnostic, if it has not already been reported. + pub fn take_message(&self) -> Option { + match self.observe() { + ReaderFailureObservation::Message(message) => Some(message), + ReaderFailureObservation::Running | ReaderFailureObservation::FailedWithoutMessage => { + None + } + } + } + + /// Observe failure state and consume its diagnostic under one lock. + pub fn observe(&self) -> ReaderFailureObservation { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(message) = state.message.take() { + ReaderFailureObservation::Message(message) + } else if state.failed { + ReaderFailureObservation::FailedWithoutMessage + } else { + ReaderFailureObservation::Running + } + } +} + +#[cfg(test)] +mod tests { + use super::{ReaderFailure, ReaderFailureObservation}; + use std::sync::{Arc, mpsc}; + + #[test] + fn failure_remains_set_after_message_is_consumed_while_reader_finishes() { + let failure = Arc::new(ReaderFailure::default()); + let reader_failure = Arc::clone(&failure); + let (published_tx, published_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let reader = std::thread::spawn(move || { + reader_failure.publish("captured reader panic".to_string()); + published_tx.send(()).expect("test waiter should remain"); + release_rx.recv().expect("test should release reader"); + }); + + published_rx.recv().expect("reader should publish failure"); + assert!(!reader.is_finished(), "reader must still be finishing"); + assert_eq!( + failure.take_message().as_deref(), + Some("captured reader panic") + ); + let observation = failure.observe(); + + release_tx.send(()).expect("reader should remain blocked"); + reader.join().expect("test reader should exit cleanly"); + + assert_eq!( + observation, + ReaderFailureObservation::FailedWithoutMessage, + "consuming the diagnostic must preserve terminal failure state" + ); + } +} diff --git a/tests/e2e-cucumber/src/send_until.rs b/tests/e2e-cucumber/src/send_until.rs new file mode 100644 index 00000000..767cb5ee --- /dev/null +++ b/tests/e2e-cucumber/src/send_until.rs @@ -0,0 +1,273 @@ +// Copyright © Advanced Micro Devices, Inc., or its affiliates. +// +// SPDX-License-Identifier: MIT + +//! Retry loop for sending an idempotent terminal input until its effect appears. + +use std::future::Future; +use std::pin::Pin; +use std::time::{Duration, Instant}; + +/// Borrowing future returned by the wait operation used by [`send_until`]. +pub type WaitFuture<'a> = Pin> + Send + 'a>>; + +/// Deadline and per-attempt wait used by [`send_until`]. +#[derive(Clone, Copy, Debug)] +pub struct RetryTiming { + pub timeout: Duration, + pub resend_interval: Duration, +} + +/// Result of checking whether a failed wait may be retried. +#[derive(Debug, PartialEq, Eq)] +pub enum TerminalState { + /// The session is still live, so another send is allowed. + Running, + /// The session stopped, and the failed wait already carries the best error. + Stopped, + /// A terminal error landed after the wait and supersedes its retryable error. + Failed(String), +} + +/// Send `bytes`, wait for `marker`, and retry until success, terminal state, or +/// the deadline. A terminal state returns the failed wait's exact error. +pub async fn send_until( + state: &mut S, + bytes: &str, + marker: &str, + timing: RetryTiming, + mut send: SendInput, + mut wait: Wait, + mut terminal: Terminal, +) -> Result<(), String> +where + S: Send, + SendInput: FnMut(&mut S, &str) -> Result<(), String> + Send, + Wait: for<'a> FnMut(&'a mut S, &'a str, Duration) -> WaitFuture<'a> + Send, + Terminal: FnMut(&mut S, &str) -> TerminalState + Send, +{ + let RetryTiming { + timeout, + resend_interval, + } = timing; + let deadline = Instant::now() + timeout; + loop { + send(state, bytes)?; + let remaining = deadline.saturating_duration_since(Instant::now()); + let attempt = resend_interval.min(remaining); + let last_error = match wait(state, marker, attempt).await { + Ok(()) => return Ok(()), + Err(error) => error, + }; + match terminal(state, marker) { + TerminalState::Running => {} + TerminalState::Stopped => return Err(last_error), + TerminalState::Failed(error) => return Err(error), + } + if Instant::now() >= deadline { + return Err(format!( + "timed out after {timeout:?} waiting for {marker:?} while repeating {bytes:?}; last attempt: {last_error}" + )); + } + } +} + +#[cfg(test)] +mod tests { + use super::{RetryTiming, TerminalState, send_until}; + use crate::reader_failure::{ReaderFailure, ReaderFailureObservation}; + use std::sync::{Arc, mpsc}; + use std::time::Duration; + + #[derive(Default)] + struct TestState { + failure: Arc, + reader_finished: bool, + sends: usize, + waits: usize, + } + + #[tokio::test(flavor = "current_thread")] + async fn captured_reader_panic_stops_retry_before_reader_thread_finishes() { + const PANIC_ERROR: &str = "captured reader panic error"; + + let mut state = TestState::default(); + state.failure.publish("captured reader panic".to_string()); + assert!(!state.reader_finished, "reader must still be finishing"); + + let error = send_until( + &mut state, + "4", + "marker", + RetryTiming { + timeout: Duration::from_millis(30), + resend_interval: Duration::from_millis(5), + }, + |state, _bytes| { + state.sends += 1; + Ok(()) + }, + |state, _marker, attempt| { + Box::pin(async move { + state.waits += 1; + if state.failure.take_message().is_some() { + Err(PANIC_ERROR.to_string()) + } else { + tokio::time::sleep(attempt).await; + Err("marker timeout".to_string()) + } + }) + }, + |state, _marker| match state.failure.observe() { + ReaderFailureObservation::Message(error) => TerminalState::Failed(error), + ReaderFailureObservation::FailedWithoutMessage => TerminalState::Stopped, + ReaderFailureObservation::Running if state.reader_finished => { + TerminalState::Stopped + } + ReaderFailureObservation::Running => TerminalState::Running, + }, + ) + .await + .expect_err("reader panic must stop the retry loop"); + + assert_eq!(error, PANIC_ERROR); + assert_eq!(state.sends, 1, "terminal failure must not resend input"); + assert_eq!(state.waits, 1, "terminal failure must not wait again"); + } + + #[tokio::test(flavor = "current_thread")] + async fn terminal_error_published_as_wait_times_out_wins_over_marker_timeout() { + const PANIC_ERROR: &str = "captured reader panic error"; + const MARKER_TIMEOUT: &str = "marker timeout"; + + let failure = Arc::new(ReaderFailure::default()); + let reader_failure = Arc::clone(&failure); + let (at_boundary_tx, at_boundary_rx) = mpsc::channel(); + let (published_tx, published_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let reader = std::thread::spawn(move || { + at_boundary_rx + .recv() + .expect("wait should reach its final terminal check"); + reader_failure.publish(PANIC_ERROR.to_string()); + published_tx.send(()).expect("test waiter should remain"); + release_rx.recv().expect("test should release reader"); + }); + + let mut state = TestState { + failure: Arc::clone(&failure), + ..TestState::default() + }; + let error = send_until( + &mut state, + "4", + "marker", + RetryTiming { + timeout: Duration::from_millis(30), + resend_interval: Duration::from_millis(5), + }, + |state, _bytes| { + state.sends += 1; + Ok(()) + }, + move |state, _marker, _attempt| { + state.waits += 1; + at_boundary_tx + .send(()) + .expect("reader should wait for the boundary"); + published_rx + .recv() + .expect("reader should publish before wait returns"); + Box::pin(async { Err(MARKER_TIMEOUT.to_string()) }) + }, + |state, _marker| match state.failure.observe() { + ReaderFailureObservation::Message(error) => TerminalState::Failed(error), + ReaderFailureObservation::FailedWithoutMessage => TerminalState::Stopped, + ReaderFailureObservation::Running if state.reader_finished => { + TerminalState::Stopped + } + ReaderFailureObservation::Running => TerminalState::Running, + }, + ) + .await + .expect_err("terminal reader error must stop the retry loop"); + + assert!(!reader.is_finished(), "reader must still be finishing"); + release_tx.send(()).expect("reader should remain blocked"); + reader.join().expect("test reader should exit cleanly"); + + assert_eq!(error, PANIC_ERROR); + assert_eq!(state.sends, 1, "terminal failure must not resend input"); + assert_eq!(state.waits, 1, "terminal failure must not wait again"); + } + + #[tokio::test(flavor = "current_thread")] + async fn terminal_error_published_between_former_take_and_check_is_preserved() { + const PANIC_ERROR: &str = "captured reader panic error"; + const MARKER_TIMEOUT: &str = "marker timeout"; + + let failure = Arc::new(ReaderFailure::default()); + let reader_failure = Arc::clone(&failure); + let (after_take_tx, after_take_rx) = mpsc::channel(); + let (published_tx, published_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let reader = std::thread::spawn(move || { + after_take_rx + .recv() + .expect("terminal check should pass the former take point"); + reader_failure.publish(PANIC_ERROR.to_string()); + published_tx.send(()).expect("test waiter should remain"); + release_rx.recv().expect("test should release reader"); + }); + + let mut state = TestState { + failure: Arc::clone(&failure), + ..TestState::default() + }; + let error = send_until( + &mut state, + "4", + "marker", + RetryTiming { + timeout: Duration::from_millis(30), + resend_interval: Duration::from_millis(5), + }, + |state, _bytes| { + state.sends += 1; + Ok(()) + }, + |state, _marker, _attempt| { + state.waits += 1; + Box::pin(async { Err(MARKER_TIMEOUT.to_string()) }) + }, + move |state, _marker| { + assert_eq!( + state.failure.take_message(), + None, + "failure must publish after the former take point" + ); + after_take_tx + .send(()) + .expect("reader should wait for the boundary"); + published_rx + .recv() + .expect("reader should publish before atomic observation"); + match state.failure.observe() { + ReaderFailureObservation::Message(error) => TerminalState::Failed(error), + ReaderFailureObservation::FailedWithoutMessage => TerminalState::Stopped, + ReaderFailureObservation::Running => TerminalState::Running, + } + }, + ) + .await + .expect_err("terminal reader error must stop the retry loop"); + + assert!(!reader.is_finished(), "reader must still be finishing"); + release_tx.send(()).expect("reader should remain blocked"); + reader.join().expect("test reader should exit cleanly"); + + assert_eq!(error, PANIC_ERROR); + assert_eq!(state.sends, 1, "terminal failure must not resend input"); + assert_eq!(state.waits, 1, "terminal failure must not wait again"); + } +} diff --git a/tests/e2e-cucumber/tests/e2e/dash_steps.rs b/tests/e2e-cucumber/tests/e2e/dash_steps.rs index e12fe396..71600ae8 100644 --- a/tests/e2e-cucumber/tests/e2e/dash_steps.rs +++ b/tests/e2e-cucumber/tests/e2e/dash_steps.rs @@ -122,7 +122,15 @@ async fn open_observe_view(world: &mut E2eWorld) { let tui = session(world); tui.use_detail_size() .unwrap_or_else(|e| panic!("failed to enlarge the dashboard: {e}")); - tui.send("4") + // Every scenario using this step launches the dashboard and comes straight + // here, with no assertion in between to prove the TUI is reading input yet + // (unlike the ROCm journey, which asserts the home view first). A key + // written that early can be swallowed before the event loop exists, so + // repeat it until the Observe tab is actually selected — the `●` marks the + // active chip. The step then fails only if the dashboard never gets there, + // not if it was slow to start. + tui.send_until("4", "● Observe", DEFAULT_TIMEOUT) + .await .unwrap_or_else(|e| panic!("failed to switch to the Observe tab: {e}")); } diff --git a/tests/e2e-cucumber/tests/e2e/tui_driver.rs b/tests/e2e-cucumber/tests/e2e/tui_driver.rs index 9996dba4..cd4fca15 100644 --- a/tests/e2e-cucumber/tests/e2e/tui_driver.rs +++ b/tests/e2e-cucumber/tests/e2e/tui_driver.rs @@ -30,6 +30,8 @@ use std::time::{Duration, Instant}; use portable_pty::{Child, CommandBuilder, MasterPty, PtySize, native_pty_system}; use e2e_cucumber::panic_capture::panic_message; +use e2e_cucumber::reader_failure::{ReaderFailure, ReaderFailureObservation}; +use e2e_cucumber::send_until::{RetryTiming, TerminalState, send_until as retry_send_until}; use crate::E2eWorld; @@ -47,6 +49,10 @@ const DETAIL_COLS: u16 = 120; /// poll cadence, not a fixed readiness sleep: every wait has a deadline and /// returns the instant its condition holds. const POLL_INTERVAL: Duration = Duration::from_millis(20); +/// How long [`TuiSession::send_until`] waits for a key to take effect before +/// sending it again. Long enough that a busy host is not spammed with repeats, +/// short enough that several attempts fit inside a normal step timeout. +const KEY_RESEND_INTERVAL: Duration = Duration::from_millis(500); /// Maximum time to let the PTY reader consume the child's final frame after the /// process exits. This is bounded so a misbehaving PTY cannot stall a scenario. const DRAIN_TIMEOUT: Duration = Duration::from_millis(250); @@ -64,12 +70,10 @@ pub struct TuiSession { parser: Arc>, reader_stop: Arc, reader: Option>, - /// Set by the reader thread if `vt100::Parser::process` ever panics, before - /// the thread exits. `wait_for_screen`/`wait_for_exit` check this every poll - /// so a reader panic (which would otherwise just stop screen updates and - /// poison `parser`) is reported directly instead of surfacing as a 30s - /// timeout over an unexplained blank/stale screen. - reader_panic: Arc>>, + /// Published by the reader thread if `vt100::Parser::process` panics. Keeps + /// terminal failure state after the one-time diagnostic is consumed, so a + /// retry cannot lose the cause while the reader thread is still finishing. + reader_failure: Arc, /// Kept alive for the lifetime of the session: the reader/writer are cloned /// from it, and dropping it early would close the PTY. master: Box, @@ -179,12 +183,12 @@ impl TuiSession { let parser = Arc::new(Mutex::new(vt100::Parser::new(ROWS, COLS, 0))); let reader_stop = Arc::new(AtomicBool::new(false)); - let reader_panic: Arc>> = Arc::new(Mutex::new(None)); + let reader_failure = Arc::new(ReaderFailure::default()); let reader = spawn_reader( reader, Arc::clone(&parser), Arc::clone(&reader_stop), - Arc::clone(&reader_panic), + Arc::clone(&reader_failure), ); Ok(Self { @@ -193,7 +197,7 @@ impl TuiSession { parser, reader_stop, reader: Some(reader), - reader_panic, + reader_failure, master: pair.master, finished: false, recorded: false, @@ -240,10 +244,7 @@ impl TuiSession { /// direct diagnostic instead of a 30s timeout over a screen that stopped /// updating for an unexplained reason. fn take_reader_panic(&self) -> Option { - self.reader_panic - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .take() + self.reader_failure.take_message() } /// Resize both the real PTY and the emulated screen. The application receives @@ -278,6 +279,79 @@ impl TuiSession { .map_err(|e| format!("failed to write to pty: {e}")) } + /// Retrieve a terminal failure that landed after a wait's final poll but + /// before the retry loop decides whether another key is safe to send. + fn terminal_state_after_wait(&mut self, marker: &str) -> TerminalState { + let reader_finished = self + .reader + .as_ref() + .is_some_and(std::thread::JoinHandle::is_finished); + let reader_failure = self.reader_failure.observe(); + if let ReaderFailureObservation::Message(panic_message) = &reader_failure { + return TerminalState::Failed(format!( + "pty reader thread panicked while waiting for {marker:?}: {panic_message}\n{}", + self.framed_screen() + )); + } + if self.finished { + return TerminalState::Stopped; + } + match self.child.try_wait() { + Ok(Some(status)) => { + self.finished = true; + self.record_once(i32::try_from(status.exit_code()).unwrap_or(-1)); + TerminalState::Failed(format!( + "process exited ({status:?}) before {marker:?} appeared.\n{}", + self.framed_screen() + )) + } + Ok(None) => match reader_failure { + ReaderFailureObservation::FailedWithoutMessage => TerminalState::Stopped, + ReaderFailureObservation::Running if reader_finished => TerminalState::Stopped, + ReaderFailureObservation::Running => TerminalState::Running, + ReaderFailureObservation::Message(_) => unreachable!("handled above"), + }, + Err(error) => TerminalState::Failed(format!("failed to poll TUI child: {error}")), + } + } + + /// Send `bytes` until the screen shows `marker`, re-sending every + /// [`KEY_RESEND_INTERVAL`] until the deadline. + /// + /// A bare [`send`](Self::send) writes into the pseudo-terminal whether or + /// not the application is reading yet, so a keystroke typed during startup + /// can be consumed by whatever holds the terminal at that moment and never + /// reach the event loop. The key is then simply lost — nothing retries it, + /// and the scenario fails much later, in an assertion about a view it never + /// left. Re-sending until the expected view appears makes the step depend on + /// the application having acted on the key rather than on it having been + /// ready when the key was written. + /// + /// Only safe for idempotent keys (a tab jump, not a toggle): the key is + /// always sent at least once, and further copies may still be queued in the + /// terminal when the marker appears, so the application may act on it again + /// after this returns. + pub async fn send_until( + &mut self, + bytes: &str, + marker: &str, + timeout: Duration, + ) -> Result<(), String> { + retry_send_until( + self, + bytes, + marker, + RetryTiming { + timeout, + resend_interval: KEY_RESEND_INTERVAL, + }, + Self::send, + |session, marker, attempt| Box::pin(session.wait_for_screen(marker, attempt)), + Self::terminal_state_after_wait, + ) + .await + } + /// Poll the current screen until it contains `marker`, or fail with a /// deadline that includes the last screen for diagnosis. Also fails fast if /// the child exits before the marker appears. @@ -434,7 +508,7 @@ impl Drop for TuiSession { self.reader_stop.store(true, Ordering::Relaxed); if let Some(handle) = self.reader.take() { // `join` returns `Err` only if the reader thread itself panicked - // (distinct from `reader_panic`, which we set *before* the thread + // (distinct from `reader_failure`, which we publish *before* the thread // exits normally after catching a `p.process` panic — so `join` // failing here would mean some other, uncaught panic in the reader). // Never re-panic here: if a scenario step already panicked and this @@ -476,14 +550,15 @@ impl Drop for TuiSession { /// inspects the resulting `Screen`, never the reader. /// /// If `vt100::Parser::process` ever panics, it's caught here (rather than left -/// to unwind the reader thread silently) and recorded into `reader_panic` before -/// the thread exits, so `wait_for_screen`/`wait_for_exit` can fail fast with the -/// actual cause instead of quietly polling a screen that will never update again. +/// to unwind the reader thread silently) and published through `reader_failure` +/// before the thread exits, so `wait_for_screen`/`wait_for_exit` can fail fast +/// with the actual cause instead of quietly polling a screen that will never +/// update again. fn spawn_reader( mut reader: Box, parser: Arc>, stop: Arc, - reader_panic: Arc>>, + reader_failure: Arc, ) -> JoinHandle<()> { std::thread::spawn(move || { let mut buf = [0u8; 8192]; @@ -503,9 +578,7 @@ fn spawn_reader( drop(p); if let Err(payload) = result { let message = panic_message(&payload); - *reader_panic - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(message); + reader_failure.publish(message); break; } }