scope-bca-cli 0.8.0

//! # Prompt input
//!
//! Reads the answer to an interactive prompt (the setup wizard, a token
//! selection) from stdin, so that a cancel can stop the read.
//!
//! In the TUI, a suspend-mode command reads its prompts on the plain
//! terminal. Ctrl-C there must cancel the command and return to the TUI. A
//! blocking `read_line` cannot be stopped, and a read on another thread
//! would stay behind and take the TUI's next keys. So the reader waits for
//! input in short steps and checks `CANCEL` between them. It reads one
//! byte at a time, so it never takes bytes after the newline.
//!
//! On Windows the console edits the line itself and hands it over whole,
//! so the reader calls `ReadConsoleW` and waits for the line. Ctrl-C makes
//! that read return empty with `ERROR_OPERATION_ABORTED` (std retries
//! then, so a `read_line` would keep waiting); the reader takes it as the
//! cancel.
//!
//! Outside the TUI nothing sets `CANCEL`, and the reader acts as
//! `read_line` does.
//!
//! Every prompt goes through `ask` (or `pick`, for a numbered list).
//! It pauses the progress bars, so a spinner never draws over a question.
//! A question, a list to pick from and the text around a prompt are not
//! data: they go to the diagnostic channel of the [`Output`]. So piped
//! data stays clean, and a TUI run keeps only data for Ctrl-Y and `save`.

use crate::cli::output::Output;
use scope::error::ScopeError;
use std::io::{self, BufRead};
use std::sync::atomic::AtomicBool;
#[cfg(any(unix, test))]
use std::sync::atomic::Ordering;

/// Set when a signal cancels the suspend-mode command. A prompt read that
/// sees it stops with [`io::ErrorKind::Interrupted`]. The TUI clears it
/// when a suspend-mode command starts.
pub(crate) static CANCEL: AtomicBool = AtomicBool::new(false);

/// How long the reader waits for input before it checks the cancel flag
/// again. It is the longest delay between a cancel and the read's return.
#[cfg(unix)]
const POLL_INTERVAL_MS: libc::c_int = 100;

/// The error a cancelled read returns.
#[cfg(any(unix, windows))]
fn cancelled() -> io::Error {
    io::Error::new(io::ErrorKind::Interrupted, "cancelled")
}

/// Writes `question` to the diagnostic channel of `out` and reads one
/// line of answer from `reader`. Every progress bar is paused from before
/// the question is written until the answer is read, so no bar draws over
/// it. The question may span lines. The channel is stderr on the CLI,
/// which Rust does not buffer, so the question shows before the read.
///
/// # Errors
///
/// Returns the write or read error.
pub(crate) fn ask(reader: &mut impl BufRead, out: &Output, question: &str) -> io::Result<String> {
    crate::cli::progress::paused(|| {
        crate::err!(out, "{question}")?;
        let mut answer = String::new();
        reader.read_line(&mut answer)?;
        Ok(answer)
    })
}

/// `ask` on stdin (cancellable in the TUI). The `scope` binary's own
/// prompts use it.
///
/// # Errors
///
/// Returns the write or read error.
pub fn ask_stdin(out: &Output, question: &str) -> io::Result<String> {
    ask(&mut stdin(), out, question)
}

/// `ask` on stdin that waits at most `timeout` for the answer. Returns
/// `None` when no answer came in time.
///
/// The read runs on a thread of its own, which stays behind, blocked on
/// stdin, when the time runs out. Use it only as the process's last
/// question (the exit question about the Ghola sidecar).
///
/// # Errors
///
/// Returns the write or read error.
pub fn ask_timeout(
    out: &Output,
    question: &str,
    timeout: std::time::Duration,
) -> io::Result<Option<String>> {
    crate::cli::progress::paused(|| {
        crate::err!(out, "{question}")?;
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let mut answer = String::new();
            let read = stdin().read_line(&mut answer).map(|_| answer);
            let _ = tx.send(read);
        });
        match rx.recv_timeout(timeout) {
            Ok(read) => read.map(Some),
            Err(_) => Ok(None),
        }
    })
}

/// Shows `list` (it ends with a newline), asks `Select <what> (1-<count>): `
/// and returns the zero-based index of the answer.
///
/// # Errors
///
/// The write or read error; `Api` when the answer is not a number
/// from 1 to `count`.
pub(crate) fn pick(
    reader: &mut impl BufRead,
    out: &Output,
    list: &str,
    what: &str,
    count: usize,
) -> scope::error::Result<usize> {
    let question = format!("{list}\nSelect {what} (1-{count}): ");
    let answer = ask(reader, out, &question)?;
    let n: usize = answer
        .trim()
        .parse()
        .map_err(|_| ScopeError::Api("Invalid selection".to_string()))?;
    if !(1..=count).contains(&n) {
        return Err(ScopeError::Api(format!(
            "Selection must be between 1 and {count}"
        )));
    }
    Ok(n - 1)
}

/// The input for a prompt: stdin, cancellable with [`CANCEL`].
#[cfg(unix)]
pub(crate) fn stdin() -> PromptInput<'static> {
    PromptInput::new(libc::STDIN_FILENO, &CANCEL)
}

/// The input for a prompt: stdin, cancellable by Ctrl-C at the console.
#[cfg(windows)]
pub(crate) fn stdin() -> console::ConsoleInput<'static> {
    console::ConsoleInput::new(&CANCEL)
}

/// The input for a prompt: stdin. There is no cancel off unix and Windows.
#[cfg(not(any(unix, windows)))]
pub(crate) fn stdin() -> io::StdinLock<'static> {
    io::stdin().lock()
}

#[cfg(windows)]
mod console {
    use super::cancelled;
    use std::io::{self, BufRead, Read};
    use std::sync::atomic::{AtomicBool, Ordering};
    use windows_sys::Win32::Foundation::{ERROR_OPERATION_ABORTED, GetLastError, HANDLE};
    use windows_sys::Win32::System::Console::{
        GetConsoleMode, GetStdHandle, ReadConsoleW, STD_INPUT_HANDLE,
    };

    /// Prompt input on Windows. A console stdin is read with
    /// `ReadConsoleW`, so Ctrl-C ends the read; any other stdin (a pipe,
    /// a file) is read as std reads it.
    pub(crate) struct ConsoleInput<'a> {
        /// The console input handle; `None` when stdin is not a console.
        console: Option<HANDLE>,
        std: io::StdinLock<'static>,
        cancel: &'a AtomicBool,
        /// Bytes of a console read that the caller has not taken yet.
        pending: Vec<u8>,
    }

    impl<'a> ConsoleInput<'a> {
        pub(crate) fn new(cancel: &'a AtomicBool) -> Self {
            // SAFETY: GetStdHandle takes no pointers; GetConsoleMode
            // writes one u32 and fails on a handle that is not a console.
            let console = unsafe {
                let handle = GetStdHandle(STD_INPUT_HANDLE);
                let mut mode = 0u32;
                (GetConsoleMode(handle, &mut mode) != 0).then_some(handle)
            };
            Self {
                console,
                std: io::stdin().lock(),
                cancel,
                pending: Vec::new(),
            }
        }

        /// Reads console input up to and including the end of a line into
        /// `pending`. Leaves it empty at end of input (Ctrl-Z).
        ///
        /// # Errors
        ///
        /// `Interrupted` when Ctrl-C aborts the read or the cancel flag is
        /// set; `InvalidData` when the input is not valid UTF-16; or the
        /// read error.
        fn read_console_line(&mut self, handle: HANDLE) -> io::Result<()> {
            let mut line: Vec<u16> = Vec::new();
            let mut buf = [0u16; 1024];
            loop {
                if self.cancel.load(Ordering::SeqCst) {
                    return Err(cancelled());
                }
                let mut read = 0u32;
                // SAFETY: the buffer is valid for `buf.len()` u16s, and
                // ReadConsoleW writes at most that many and the count.
                let ok = unsafe {
                    ReadConsoleW(
                        handle,
                        buf.as_mut_ptr().cast(),
                        buf.len() as u32,
                        &mut read,
                        std::ptr::null(),
                    )
                };
                if ok == 0 {
                    return Err(io::Error::last_os_error());
                }
                if read == 0 {
                    // SAFETY: reads the calling thread's last error.
                    if unsafe { GetLastError() } == ERROR_OPERATION_ABORTED {
                        // Ctrl-C. The TUI's handler sets the flag too, but
                        // maybe not yet, and run_suspended checks it.
                        self.cancel.store(true, Ordering::SeqCst);
                        return Err(cancelled());
                    }
                    break;
                }
                let got = &buf[..read as usize];
                line.extend_from_slice(got);
                // Ctrl-Z at the start of a line is end of input, as std
                // reads it.
                if line.first() == Some(&0x1a) {
                    line.clear();
                    break;
                }
                if got.last() == Some(&u16::from(b'\n')) {
                    break;
                }
            }
            let text = String::from_utf16(&line).map_err(|_| {
                io::Error::new(
                    io::ErrorKind::InvalidData,
                    "console input was not valid UTF-16",
                )
            })?;
            self.pending = text.into_bytes();
            Ok(())
        }
    }

    impl Read for ConsoleInput<'_> {
        fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
            let available = self.fill_buf()?;
            let n = available.len().min(out.len());
            out[..n].copy_from_slice(&available[..n]);
            self.consume(n);
            Ok(n)
        }
    }

    impl BufRead for ConsoleInput<'_> {
        /// A cancel returns `Other`, not `Interrupted`: the std helpers
        /// over `fill_buf` retry on `Interrupted`, so they would never
        /// stop.
        fn fill_buf(&mut self) -> io::Result<&[u8]> {
            let Some(handle) = self.console else {
                return self.std.fill_buf();
            };
            if self.pending.is_empty() {
                match self.read_console_line(handle) {
                    Err(e) if e.kind() == io::ErrorKind::Interrupted => {
                        return Err(io::Error::other(e));
                    }
                    other => other?,
                }
            }
            Ok(&self.pending)
        }

        fn consume(&mut self, amt: usize) {
            if self.console.is_none() {
                self.std.consume(amt);
            } else {
                self.pending.drain(..amt.min(self.pending.len()));
            }
        }

        /// Reads one line, as `BufRead::read_line` does, and appends it
        /// to `buf`. Returns the number of bytes read; 0 at end of input.
        ///
        /// # Errors
        ///
        /// `Interrupted` when Ctrl-C cancels the read; or the read error.
        fn read_line(&mut self, buf: &mut String) -> io::Result<usize> {
            let Some(handle) = self.console else {
                return self.std.read_line(buf);
            };
            if self.pending.is_empty() {
                self.read_console_line(handle)?;
            }
            let end = self
                .pending
                .iter()
                .position(|&b| b == b'\n')
                .map_or(self.pending.len(), |i| i + 1);
            let line: Vec<u8> = self.pending.drain(..end).collect();
            let text = String::from_utf8(line).expect("decoded from UTF-16");
            buf.push_str(&text);
            Ok(text.len())
        }
    }
}

/// A line reader over a file descriptor that stops when `cancel` is set.
#[cfg(unix)]
pub(crate) struct PromptInput<'a> {
    fd: std::os::fd::RawFd,
    cancel: &'a AtomicBool,
    /// The byte `fill_buf` read and `consume` did not take yet.
    byte: [u8; 1],
    has_byte: bool,
}

#[cfg(unix)]
impl<'a> PromptInput<'a> {
    pub(crate) fn new(fd: std::os::fd::RawFd, cancel: &'a AtomicBool) -> Self {
        Self {
            fd,
            cancel,
            byte: [0],
            has_byte: false,
        }
    }

    /// Reads one byte. `None` at end of input. Waits in steps of
    /// [`POLL_INTERVAL_MS`] and stops with `Interrupted` when `cancel` is
    /// set.
    fn read_byte(&mut self) -> io::Result<Option<u8>> {
        if std::mem::take(&mut self.has_byte) {
            return Ok(Some(self.byte[0]));
        }
        loop {
            if self.cancel.load(Ordering::SeqCst) {
                return Err(cancelled());
            }
            let mut pfd = libc::pollfd {
                fd: self.fd,
                events: libc::POLLIN,
                revents: 0,
            };
            // SAFETY: one valid pollfd, and poll writes only its revents.
            let ready = unsafe { libc::poll(&mut pfd, 1, POLL_INTERVAL_MS) };
            if ready < 0 {
                let e = io::Error::last_os_error();
                if e.kind() == io::ErrorKind::Interrupted {
                    continue;
                }
                return Err(e);
            }
            if ready == 0 {
                continue;
            }
            // Input, end of input or an error: the read tells which.
            let mut b = 0u8;
            // SAFETY: reads at most one byte into a valid one-byte buffer.
            let n = unsafe { libc::read(self.fd, (&mut b as *mut u8).cast(), 1) };
            match n {
                1 => return Ok(Some(b)),
                0 => return Ok(None),
                _ => {
                    let e = io::Error::last_os_error();
                    if matches!(
                        e.kind(),
                        io::ErrorKind::Interrupted | io::ErrorKind::WouldBlock
                    ) {
                        continue;
                    }
                    return Err(e);
                }
            }
        }
    }
}

#[cfg(unix)]
impl io::Read for PromptInput<'_> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let Some(first) = buf.first_mut() else {
            return Ok(0);
        };
        Ok(match self.read_byte()? {
            Some(b) => {
                *first = b;
                1
            }
            None => 0,
        })
    }
}

#[cfg(unix)]
impl BufRead for PromptInput<'_> {
    /// The prompts use `read_line`; this exists because `BufRead` needs
    /// it. A cancel returns `Other`, not `Interrupted`: the std helpers
    /// over `fill_buf` retry on `Interrupted`, so they would never stop.
    fn fill_buf(&mut self) -> io::Result<&[u8]> {
        if !self.has_byte {
            match self.read_byte() {
                Ok(Some(b)) => {
                    self.byte[0] = b;
                    self.has_byte = true;
                }
                Ok(None) => {}
                Err(e) if e.kind() == io::ErrorKind::Interrupted => {
                    return Err(io::Error::other(e));
                }
                Err(e) => return Err(e),
            }
        }
        Ok(&self.byte[..usize::from(self.has_byte)])
    }

    fn consume(&mut self, amt: usize) {
        if amt > 0 {
            self.has_byte = false;
        }
    }

    /// Reads up to and including the newline, as `BufRead::read_line`
    /// does, and appends it to `buf`. Returns the number of bytes read; 0
    /// at end of input.
    ///
    /// # Errors
    ///
    /// `Interrupted` when the cancel flag is set; `InvalidData` when the
    /// line is not UTF-8 (`buf` is then unchanged); or the read error.
    fn read_line(&mut self, buf: &mut String) -> io::Result<usize> {
        let mut line = Vec::new();
        while let Some(b) = self.read_byte()? {
            line.push(b);
            if b == b'\n' {
                break;
            }
        }
        let text = String::from_utf8(line).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidData,
                "stream did not contain valid UTF-8",
            )
        })?;
        buf.push_str(&text);
        Ok(text.len())
    }
}

#[cfg(test)]
mod ask_tests {
    use super::*;
    use crate::cli::progress::testing::{REGISTRY_LOCK, Screen, drawing_spinner};
    use std::io::Write;
    use std::sync::Arc;
    use std::sync::atomic::AtomicUsize;
    use std::time::Duration;

    /// An answer that takes a while to type. It notes how many draws the
    /// screen had when the answer came.
    struct SlowAnswer<'a> {
        screen: &'a Screen,
        draws_at_answer: Option<usize>,
    }

    impl io::Read for SlowAnswer<'_> {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            if self.draws_at_answer.is_some() {
                return Ok(0);
            }
            std::thread::sleep(Duration::from_millis(40));
            self.draws_at_answer = Some(self.screen.ops().len());
            buf[..2].copy_from_slice(b"y\n");
            Ok(2)
        }
    }

    /// A diagnostic channel that notes how many draws the screen had when
    /// the question was written.
    struct Question {
        screen: Screen,
        draws_at_question: Arc<AtomicUsize>,
    }

    impl Write for Question {
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            self.draws_at_question
                .store(self.screen.ops().len(), Ordering::SeqCst);
            Ok(buf.len())
        }
        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    #[test]
    fn test_ask_keeps_a_ticking_spinner_off_the_question() {
        // crawl by symbol: the "Searching..." spinner drew over "Save this
        // token for future use? [y/N]" until the answer came.
        let _lock = REGISTRY_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let screen = Screen::default();
        let sp = drawing_spinner("Searching...", &Output::capture().0, &screen);
        std::thread::sleep(Duration::from_millis(30));
        let mut answer = SlowAnswer {
            screen: &screen,
            draws_at_answer: None,
        };
        let mut reader = io::BufReader::new(&mut answer);
        let draws_at_question = Arc::new(AtomicUsize::new(0));
        let out = Output::from_writers(
            Vec::new(),
            Question {
                screen: screen.clone(),
                draws_at_question: draws_at_question.clone(),
            },
        );
        let got = ask(&mut reader, &out, "Save? [y/N]: ").unwrap();
        drop(reader);
        assert_eq!(got, "y\n");
        assert_eq!(
            answer.draws_at_answer,
            Some(draws_at_question.load(Ordering::SeqCst)),
            "the spinner drew between the question and the answer"
        );
        sp.finish_and_clear();
    }

    #[test]
    fn test_a_prompt_never_reaches_the_data_channel() {
        // `scope crawl USDC > out.json`: the question must show on the
        // terminal and stay out of the file.
        let (out, cap) = Output::capture();
        let got = ask(&mut io::Cursor::new(b"y\n".to_vec()), &out, "Save? [y/N]: ").unwrap();
        assert_eq!(got, "y\n");
        assert_eq!(cap.out(), "");
        assert_eq!(cap.err(), "Save? [y/N]: ");
    }

    #[test]
    fn test_ask_returns_the_write_error() {
        let out = Output::from_writers(Vec::new(), crate::cli::output::tests::ClosedPipe);
        let e = ask(&mut io::Cursor::new(b"y\n".to_vec()), &out, "q? ").unwrap_err();
        assert_eq!(e.kind(), io::ErrorKind::BrokenPipe);
    }

    #[test]
    fn test_pick_returns_the_index_and_rejects_bad_answers() {
        let pick_with = |answer: &str| {
            let (out, cap) = Output::capture();
            let got = pick(
                &mut io::Cursor::new(answer.as_bytes().to_vec()),
                &out,
                "1 a\n2 b\n",
                "token",
                2,
            );
            assert_eq!(cap.err(), "1 a\n2 b\n\nSelect token (1-2): ");
            assert_eq!(cap.out(), "", "the list is not data");
            got.map_err(|e| e.to_string())
        };
        assert_eq!(pick_with("2\n"), Ok(1));
        assert!(pick_with("x\n").unwrap_err().contains("Invalid selection"));
        assert!(
            pick_with("3\n")
                .unwrap_err()
                .contains("Selection must be between 1 and 2")
        );
        assert!(
            pick_with("0\n")
                .unwrap_err()
                .contains("Selection must be between 1 and 2")
        );
    }
}

#[cfg(all(test, unix))]
mod tests {
    use super::*;
    use std::io::Read;
    use std::os::fd::RawFd;
    use std::time::{Duration, Instant};

    /// A pipe: (read end, write end). Closed on drop.
    struct Pipe(RawFd, RawFd);

    impl Pipe {
        fn new() -> Self {
            let mut fds = [0; 2];
            // SAFETY: pipe writes two fds into a valid two-element array.
            assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0);
            Pipe(fds[0], fds[1])
        }

        fn write(&self, bytes: &[u8]) {
            // SAFETY: writes from a valid buffer of the given length.
            let n = unsafe { libc::write(self.1, bytes.as_ptr().cast(), bytes.len()) };
            assert_eq!(n, bytes.len() as isize);
        }

        fn close_write(&mut self) {
            // SAFETY: closes the write end once; -1 marks it closed.
            unsafe { libc::close(self.1) };
            self.1 = -1;
        }

        /// What is left in the pipe, read without the prompt reader.
        fn rest(&mut self) -> Vec<u8> {
            self.close_write();
            let mut out = Vec::new();
            let mut b = [0u8; 64];
            loop {
                // SAFETY: reads into a valid buffer of the given length.
                let n = unsafe { libc::read(self.0, b.as_mut_ptr().cast(), b.len()) };
                if n <= 0 {
                    return out;
                }
                out.extend_from_slice(&b[..n as usize]);
            }
        }
    }

    impl Drop for Pipe {
        fn drop(&mut self) {
            for fd in [self.0, self.1] {
                if fd >= 0 {
                    // SAFETY: each open fd is closed once.
                    unsafe { libc::close(fd) };
                }
            }
        }
    }

    #[test]
    fn test_reads_one_line_and_leaves_the_rest() {
        // The next prompt reads the next line: the reader must not take
        // bytes after the newline.
        let flag = AtomicBool::new(false);
        let mut pipe = Pipe::new();
        pipe.write(b"yes\nnext\n");
        let mut input = PromptInput::new(pipe.0, &flag);
        let mut line = String::from("old:");
        assert_eq!(input.read_line(&mut line).unwrap(), 4);
        assert_eq!(line, "old:yes\n", "appends, as read_line does");
        assert_eq!(pipe.rest(), b"next\n");
    }

    #[test]
    fn test_end_of_input_returns_the_partial_line_then_zero() {
        // A closed stdin must end the prompt as read_line does, not hang.
        let flag = AtomicBool::new(false);
        let mut pipe = Pipe::new();
        pipe.write(b"abc");
        pipe.close_write();
        let mut input = PromptInput::new(pipe.0, &flag);
        let mut line = String::new();
        assert_eq!(input.read_line(&mut line).unwrap(), 3);
        assert_eq!(line, "abc");
        assert_eq!(input.read_line(&mut line).unwrap(), 0);
    }

    #[test]
    fn test_invalid_utf8_is_an_error_and_leaves_the_buffer() {
        let flag = AtomicBool::new(false);
        let pipe = Pipe::new();
        pipe.write(b"\xff\n");
        let mut input = PromptInput::new(pipe.0, &flag);
        let mut line = String::from("kept");
        let e = input.read_line(&mut line).unwrap_err();
        assert_eq!(e.kind(), io::ErrorKind::InvalidData);
        assert_eq!(line, "kept");
    }

    #[test]
    fn test_cancel_stops_a_waiting_read_and_takes_no_later_input() {
        // Ctrl-C at a prompt in the TUI: the read must stop soon, and no
        // reader may stay behind to take the TUI's next keys.
        let flag = AtomicBool::new(false);
        let mut pipe = Pipe::new();
        let started = Instant::now();
        let e = std::thread::scope(|s| {
            s.spawn(|| {
                std::thread::sleep(Duration::from_millis(150));
                flag.store(true, Ordering::SeqCst);
            });
            let mut line = String::new();
            PromptInput::new(pipe.0, &flag)
                .read_line(&mut line)
                .unwrap_err()
        });
        assert_eq!(e.kind(), io::ErrorKind::Interrupted);
        assert!(started.elapsed() < Duration::from_secs(1), "stopped late");
        pipe.write(b"k");
        std::thread::sleep(Duration::from_millis(2 * POLL_INTERVAL_MS as u64));
        assert_eq!(pipe.rest(), b"k", "the key after the cancel is still there");
    }

    #[test]
    fn test_a_set_flag_stops_the_read_before_it_waits() {
        let flag = AtomicBool::new(true);
        let mut pipe = Pipe::new();
        pipe.write(b"y\n");
        let mut line = String::new();
        let e = PromptInput::new(pipe.0, &flag)
            .read_line(&mut line)
            .unwrap_err();
        assert_eq!(e.kind(), io::ErrorKind::Interrupted);
        assert_eq!(pipe.rest(), b"y\n");
    }

    #[test]
    fn test_read_and_fill_buf_take_one_byte_at_a_time() {
        let flag = AtomicBool::new(false);
        let mut pipe = Pipe::new();
        pipe.write(b"ab");
        let mut input = PromptInput::new(pipe.0, &flag);
        assert_eq!(input.read(&mut []).unwrap(), 0);
        assert_eq!(input.fill_buf().unwrap(), b"a");
        assert_eq!(input.fill_buf().unwrap(), b"a", "not consumed yet");
        let mut buf = [0u8; 4];
        assert_eq!(input.read(&mut buf).unwrap(), 1, "the buffered byte");
        assert_eq!(buf[0], b'a');
        input.consume(0);
        assert_eq!(input.fill_buf().unwrap(), b"b");
        input.consume(1);
        assert_eq!(pipe.rest(), b"");
        assert_eq!(input.fill_buf().unwrap(), b"", "end of input");
        assert_eq!(input.read(&mut buf).unwrap(), 0);
    }

    #[test]
    fn test_fill_buf_cancel_is_not_interrupted() {
        // std's read_until retries Interrupted forever; fill_buf must not
        // return it.
        let flag = AtomicBool::new(true);
        let pipe = Pipe::new();
        let mut input = PromptInput::new(pipe.0, &flag);
        let e = input.fill_buf().unwrap_err();
        assert_eq!(e.kind(), io::ErrorKind::Other);
        let mut v = Vec::new();
        assert!(input.read_until(b'\n', &mut v).is_err(), "no endless retry");
    }

    #[test]
    fn test_a_read_error_is_returned() {
        // A directory polls as ready, and reading it fails (EISDIR).
        let flag = AtomicBool::new(false);
        // SAFETY: a valid NUL-terminated path; the fd is closed below.
        let fd = unsafe { libc::open(c"/".as_ptr(), libc::O_RDONLY) };
        assert!(fd >= 0);
        let mut line = String::new();
        let r = PromptInput::new(fd, &flag).read_line(&mut line);
        // SAFETY: closes the fd opened above, once.
        unsafe { libc::close(fd) };
        assert!(r.is_err());
        assert_ne!(r.unwrap_err().kind(), io::ErrorKind::Interrupted);
    }

    #[test]
    fn test_stdin_reads_from_the_process_cancel_flag() {
        let input = stdin();
        assert_eq!(input.fd, libc::STDIN_FILENO);
        assert!(std::ptr::eq(input.cancel, &CANCEL));
    }
    #[test]
    fn test_fill_buf_returns_a_read_error_as_it_is() {
        // Only a cancel becomes `Other`; EISDIR comes back unchanged.
        let flag = AtomicBool::new(false);
        // SAFETY: a valid NUL-terminated path; the fd is closed below.
        let fd = unsafe { libc::open(c"/".as_ptr(), libc::O_RDONLY) };
        assert!(fd >= 0);
        let r = PromptInput::new(fd, &flag).fill_buf().map(<[u8]>::to_vec);
        // SAFETY: closes the fd opened above, once.
        unsafe { libc::close(fd) };
        assert_eq!(r.unwrap_err().raw_os_error(), Some(libc::EISDIR));
    }

    extern "C" fn ignore_signal(_: libc::c_int) {}

    #[test]
    fn test_a_signal_during_the_wait_does_not_end_the_read() {
        // A signal (SIGWINCH from a resize, say) makes poll fail with
        // EINTR. The reader must keep waiting, not give up the prompt.
        // SAFETY: installs a handler that does nothing, without
        // SA_RESTART, so a blocked poll returns EINTR.
        unsafe {
            let mut action: libc::sigaction = std::mem::zeroed();
            action.sa_sigaction = ignore_signal as extern "C" fn(libc::c_int) as usize;
            libc::sigemptyset(&mut action.sa_mask);
            assert_eq!(
                libc::sigaction(libc::SIGUSR1, &action, std::ptr::null_mut()),
                0
            );
        }
        let flag = AtomicBool::new(false);
        let pipe = Pipe::new();
        let fd = pipe.0;
        let (tx, rx) = std::sync::mpsc::channel();
        let line = std::thread::scope(|s| {
            let reader = s.spawn(|| {
                // SAFETY: pthread_self has no preconditions.
                tx.send(unsafe { libc::pthread_self() }).unwrap();
                let mut line = String::new();
                PromptInput::new(fd, &flag)
                    .read_line(&mut line)
                    .map(|_| line)
            });
            let thread = rx.recv().unwrap();
            for _ in 0..10 {
                std::thread::sleep(Duration::from_millis(20));
                // SAFETY: the reader thread is alive until it reads the
                // line, which is written only after the last signal.
                assert_eq!(unsafe { libc::pthread_kill(thread, libc::SIGUSR1) }, 0);
            }
            pipe.write(b"still here\n");
            reader.join().unwrap()
        });
        assert_eq!(line.unwrap(), "still here\n");
    }

    /// Puts a pipe on fd 0 for the test and puts the real stdin back on
    /// drop. Holds `DIAG_LOCK`, so no other test reads stdin meanwhile.
    struct PipedStdin {
        pipe: Pipe,
        saved: RawFd,
        _lock: tokio::sync::MutexGuard<'static, ()>,
    }

    impl PipedStdin {
        fn new() -> Self {
            let lock = crate::cli::tui::DIAG_LOCK.blocking_lock();
            let pipe = Pipe::new();
            // SAFETY: dup and dup2 on open fds; `saved` is restored and
            // closed on drop.
            let saved = unsafe { libc::dup(libc::STDIN_FILENO) };
            assert!(saved >= 0);
            assert_eq!(unsafe { libc::dup2(pipe.0, libc::STDIN_FILENO) }, 0);
            Self {
                pipe,
                saved,
                _lock: lock,
            }
        }

        /// Bytes still waiting in the pipe.
        fn unread(&self) -> usize {
            let mut n: libc::c_int = 0;
            // SAFETY: FIONREAD writes one c_int.
            assert_eq!(
                unsafe { libc::ioctl(self.pipe.0, libc::FIONREAD, &mut n) },
                0
            );
            n as usize
        }
    }

    impl Drop for PipedStdin {
        fn drop(&mut self) {
            // SAFETY: restores the stdin saved in `new` and closes the copy.
            unsafe {
                libc::dup2(self.saved, libc::STDIN_FILENO);
                libc::close(self.saved);
            }
        }
    }

    #[test]
    fn test_ask_stdin_asks_on_the_diagnostic_channel_and_reads_stdin() {
        let stdin = PipedStdin::new();
        stdin.pipe.write(b"yes\nnext\n");
        let (out, cap) = Output::capture();
        assert_eq!(ask_stdin(&out, "Continue? ").unwrap(), "yes\n");
        assert_eq!(cap.err(), "Continue? ");
        assert_eq!(cap.out(), "");
        assert_eq!(
            stdin.unread(),
            5,
            "the next line is left for the next prompt"
        );
    }

    #[test]
    fn test_ask_timeout_returns_the_answer_or_none_when_time_runs_out() {
        let stdin = PipedStdin::new();
        let (out, cap) = Output::capture();

        stdin.pipe.write(b"n\n");
        let got = ask_timeout(&out, "Stop Ghola? ", Duration::from_secs(5)).unwrap();
        assert_eq!(got.as_deref(), Some("n\n"));
        assert_eq!(cap.err(), "Stop Ghola? ");

        let started = Instant::now();
        let got = ask_timeout(&out, "Again? ", Duration::from_millis(50)).unwrap();
        assert_eq!(got, None);
        assert!(
            started.elapsed() < Duration::from_secs(2),
            "waited too long"
        );
        assert_eq!(cap.err(), "Stop Ghola? Again? ");

        // The read left behind takes this line; wait for it before the
        // real stdin goes back on fd 0.
        stdin.pipe.write(b"late\n");
        let deadline = Instant::now() + Duration::from_secs(5);
        while stdin.unread() > 0 && Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(10));
        }
        assert_eq!(stdin.unread(), 0);
    }
}