8129528574

Committed 03 Mar 2024 10:26AM UTC coverage: 47.348% (-9.2%) from 56.575%

Build # 8129528574

Build Type

Pull #68

github

Committed by

web-flow

Commit Message

Merge 43d30f6c4 into e8b43ff1b

Pull Request Pull Request #68: [WIP] Move to trait based version `Expect`

Run Details

159 of 380 new or added lines in 15 files covered. (41.84%)

724 existing lines in 16 files now uncovered.

1491 of 3149 relevant lines covered (47.35%)

3.2 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/session/sync_session.rs

//! Module contains a Session structure.

use std::{
    io::{self, BufRead, BufReader, Read, Write},
    time::{self, Duration},
};

use crate::{
    error::Error,
    expect::Expect,
    needle::Needle,
    process::{Healthcheck, NonBlocking, Termios},
    Captures,
};

/// Session represents a spawned process and its streams.
/// It controlls process and communication with it.
#[derive(Debug)]
pub struct Session<P, S> {
    proc: P,
    stream: TryStream<S>,
    expect_timeout: Option<Duration>,
    expect_lazy: bool,
}

impl<P, S> Session<P, S>
where
    S: Read,
{
    /// Creates a new session.
    pub fn new(process: P, stream: S) -> io::Result<Self> {
        let stream = TryStream::new(stream)?;

        Ok(Self {
            proc: process,
            stream,
            expect_timeout: Some(Duration::from_millis(10000)),
            expect_lazy: false,
        })
    }

    pub(crate) fn swap_stream<F, R>(mut self, new: F) -> Result<Session<P, R>, Error>
    where
        F: FnOnce(S) -> R,
        R: Read,
    {
        self.stream.flush_in_buffer();
        let buf = self.stream.get_available().to_owned();

        let stream = self.stream.into_inner();
        let stream = new(stream);

        let mut session = Session::new(self.proc, stream)?;
        session.stream.keep_in_buffer(&buf);

        Ok(session)
    }
}

impl<P, S> Session<P, S> {
    /// Set the pty session's expect timeout.
    pub fn set_expect_timeout(&mut self, expect_timeout: Option<Duration>) {
        self.expect_timeout = expect_timeout;
    }

    /// Set a expect algorithm to be either gready or lazy.
    ///
    /// Default algorithm is gready.
    ///
    /// See [Session::expect].
    pub fn set_expect_lazy(&mut self, lazy: bool) {
        self.expect_lazy = lazy;
    }

    /// Get a reference to original stream.
    pub fn get_stream(&self) -> &S {
        self.stream.as_ref()
    }

    /// Get a mut reference to original stream.
    pub fn get_stream_mut(&mut self) -> &mut S {
        self.stream.as_mut()
    }

    /// Get a reference to a process running program.
    pub fn get_process(&self) -> &P {
        &self.proc
    }

    /// Get a mut reference to a process running program.
    pub fn get_process_mut(&mut self) -> &mut P {
        &mut self.proc
    }
}

impl<P, S> Expect for Session<P, S>
where
    S: Write + Read + NonBlocking,
{
    /// Expect waits until a pattern is matched.
    ///
    /// If the method returns [Ok] it is guaranteed that at least 1 match was found.
    ///
    /// The match algorthm can be either
    ///     - gready
    ///     - lazy
    ///
    /// You can set one via [Session::set_expect_lazy].
    /// Default version is gready.
    ///
    /// The implications are.
    /// Imagine you use [crate::Regex] `"\d+"` to find a match.
    /// And your process outputs `123`.
    /// In case of lazy approach we will match `1`.
    /// Where's in case of gready one we will match `123`.
    ///
    /// # Example
    ///
    #[cfg_attr(windows, doc = "```no_run")]
    #[cfg_attr(unix, doc = "```")]
    /// let mut p = expectrl::spawn("echo 123").unwrap();
    /// let m = p.expect(expectrl::Regex("\\d+")).unwrap();
    /// assert_eq!(m.get(0).unwrap(), b"123");
    /// ```
    ///
    #[cfg_attr(windows, doc = "```no_run")]
    #[cfg_attr(unix, doc = "```")]
    /// let mut p = expectrl::spawn("echo 123").unwrap();
    /// p.set_expect_lazy(true);
    /// let m = p.expect(expectrl::Regex("\\d+")).unwrap();
    /// assert_eq!(m.get(0).unwrap(), b"1");
    /// ```
    ///
    /// This behaviour is different from [Session::check].
    ///
    /// It returns an error if timeout is reached.
    /// You can specify a timeout value by [Session::set_expect_timeout] method.
    fn expect<N>(&mut self, needle: N) -> Result<Captures, Error>
    where
        N: Needle,
    {
        match self.expect_lazy {
            true => self.expect_lazy(needle),
            false => self.expect_gready(needle),
        }
    }

    /// Check verifies if a pattern is matched.
    /// Returns empty found structure if nothing found.
    ///
    /// Is a non blocking version of [Session::expect].
    /// But its strategy of matching is different from it.
    /// It makes search against all bytes available.
    ///
    /// # Example
    ///
    #[cfg_attr(any(windows, target_os = "macos"), doc = "```no_run")]
    #[cfg_attr(not(any(target_os = "macos", windows)), doc = "```")]
    /// use expectrl::{spawn, Regex};
    /// use std::time::Duration;
    ///
    /// let mut p = spawn("echo 123").unwrap();
    /// #
    /// # // wait to guarantee that check echo worked out (most likely)
    /// # std::thread::sleep(Duration::from_millis(500));
    /// #
    /// let m = p.check(Regex("\\d+")).unwrap();
    /// assert_eq!(m.get(0).unwrap(), b"123");
    /// ```
    fn check<N>(&mut self, needle: N) -> Result<Captures, Error>
    where
        N: Needle,
    {
        let eof = self.stream.read_available()?;
        let buf = self.stream.get_available();

        let found = needle.check(buf, eof)?;
        if !found.is_empty() {
            let end_index = Captures::right_most_index(&found);
            let involved_bytes = buf[..end_index].to_vec();
            self.stream.consume_available(end_index);
            return Ok(Captures::new(involved_bytes, found));
        }

        if eof {
            return Err(Error::Eof);
        }

        Ok(Captures::new(Vec::new(), Vec::new()))
    }

    /// The functions checks if a pattern is matched.
    /// It doesn’t consumes bytes from stream.
    ///
    /// Its strategy of matching is different from the one in [Session::expect].
    /// It makes search agains all bytes available.
    ///
    /// If you want to get a matched result [Session::check] and [Session::expect] is a better option.
    /// Because it is not guaranteed that [Session::check] or [Session::expect] with the same parameters:
    ///     - will successed even right after Session::is_matched call.
    ///     - will operate on the same bytes.
    ///
    /// IMPORTANT:
    ///  
    /// If you call this method with [crate::Eof] pattern be aware that eof
    /// indication MAY be lost on the next interactions.
    /// It depends from a process you spawn.
    /// So it might be better to use [Session::check] or [Session::expect] with Eof.
    ///
    /// # Example
    ///
    #[cfg_attr(windows, doc = "```no_run")]
    #[cfg_attr(unix, doc = "```")]
    /// use expectrl::{spawn, Regex};
    /// use std::time::Duration;
    ///
    /// let mut p = spawn("cat").unwrap();
    /// p.send_line("123");
    /// # // wait to guarantee that check echo worked out (most likely)
    /// # std::thread::sleep(Duration::from_secs(1));
    /// let m = p.is_matched(Regex("\\d+")).unwrap();
    /// assert_eq!(m, true);
    /// ```
    fn is_matched<N>(&mut self, needle: N) -> Result<bool, Error>
    where
        N: Needle,
    {
        let eof = self.stream.read_available()?;
        let buf = self.stream.get_available();

        let found = needle.check(buf, eof)?;
        if !found.is_empty() {
            return Ok(true);
        }

        if eof {
            return Err(Error::Eof);
        }

        Ok(false)
    }

    /// Send text to child’s STDIN.
    ///
    /// You can also use methods from [std::io::Write] instead.
    ///
    /// # Example
    ///
    /// ```
    /// use expectrl::{spawn, ControlCode};
    ///
    /// let mut proc = spawn("cat").unwrap();
    ///
    /// proc.send("Hello");
    /// proc.send(b"World");
    /// proc.send(ControlCode::try_from("^C").unwrap());
    /// ```
    fn send<B>(&mut self, buf: B) -> Result<(), Error>
    where
        B: AsRef<[u8]>,
    {
        self.stream.write_all(buf.as_ref())?;

        Ok(())
    }

    /// Send a line to child’s STDIN.
    ///
    /// # Example
    ///
    /// ```
    /// use expectrl::{spawn, ControlCode};
    ///
    /// let mut proc = spawn("cat").unwrap();
    ///
    /// proc.send_line("Hello");
    /// proc.send_line(b"World");
    /// proc.send_line(ControlCode::try_from("^C").unwrap());
    /// ```
    fn send_line<B>(&mut self, buf: B) -> Result<(), Error>
    where
        B: AsRef<[u8]>,
    {
        #[cfg(windows)]
        const LINE_ENDING: &[u8] = b"\r\n";
        #[cfg(not(windows))]
        const LINE_ENDING: &[u8] = b"\n";

        self.stream.write_all(buf.as_ref())?;
        self.write_all(LINE_ENDING)?;

        Ok(())
    }
}

impl<P, S> Session<P, S>
where
    S: Read + NonBlocking,
{
    /// Expect which fills as much as possible to the buffer.
    ///
    /// See [Session::expect].
    fn expect_gready<N>(&mut self, needle: N) -> Result<Captures, Error>
    where
        N: Needle,
    {
        let start = time::Instant::now();
        loop {
            let eof = self.stream.read_available()?;
            let data = self.stream.get_available();

            let found = needle.check(data, eof)?;
            if !found.is_empty() {
                let end_index = Captures::right_most_index(&found);
                let involved_bytes = data[..end_index].to_vec();
                self.stream.consume_available(end_index);

                return Ok(Captures::new(involved_bytes, found));
            }

            if eof {
                return Err(Error::Eof);
            }

            if let Some(timeout) = self.expect_timeout {
                if start.elapsed() > timeout {
                    return Err(Error::ExpectTimeout);
                }
            }
        }
    }

    /// Expect which reads byte by byte.
    ///
    /// See [Session::expect].
    fn expect_lazy<N>(&mut self, needle: N) -> Result<Captures, Error>
    where
        N: Needle,
    {
        let mut checking_data_length = 0;
        let mut eof = false;
        let start = time::Instant::now();
        loop {
            let mut available = self.stream.get_available();
            if checking_data_length == available.len() {
                // We read by byte to make things as lazy as possible.
                //
                // It's chose is important in using Regex as a Needle.
                // Imagine we have a `\d+` regex.
                // Using such buffer will match string `2` imidiately eventhough right after might be other digit.
                //
                // The second reason is
                // if we wouldn't read by byte EOF indication could be lost.
                // And next blocking std::io::Read operation could be blocked forever.
                //
                // We could read all data available via `read_available` to reduce IO operations,
                // but in such case we would need to keep a EOF indicator internally in stream,
                // which is OK if EOF happens onces, but I am not sure if this is a case.
                eof = self.stream.read_available_once(&mut [0; 1])? == Some(0);
                available = self.stream.get_available();
            }

            // We intentinally not increase the counter
            // and run check one more time even though the data isn't changed.
            // Because it may be important for custom implementations of Needle.
            if checking_data_length < available.len() {
                checking_data_length += 1;
            }

            let data = &available[..checking_data_length];

            let found = needle.check(data, eof)?;
            if !found.is_empty() {
                let end_index = Captures::right_most_index(&found);
                let involved_bytes = data[..end_index].to_vec();
                self.stream.consume_available(end_index);
                return Ok(Captures::new(involved_bytes, found));
            }

            if eof {
                return Err(Error::Eof);
            }

            if let Some(timeout) = self.expect_timeout {
                if start.elapsed() > timeout {
                    return Err(Error::ExpectTimeout);
                }
            }
        }
    }
}

impl<P, S> Session<P, S>
where
    S: Read + NonBlocking,
{
    /// Try to read in a non-blocking mode.
    ///
    /// Returns [`std::io::ErrorKind::WouldBlock`]
    /// in case there's nothing to read.
    pub fn try_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.stream.try_read(buf)
    }

    /// Verifyes if stream is empty or not.
    pub fn is_empty(&mut self) -> io::Result<bool> {
        self.stream.is_empty()
    }
}

impl<P, S> Write for Session<P, S>
where
    S: Write,
{
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.stream.write(buf)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.stream.flush()
    }

    fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
        self.stream.write_vectored(bufs)
    }
}

impl<P, S> Read for Session<P, S>
where
    S: Read,
{
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.stream.read(buf)
    }
}

impl<P, S> BufRead for Session<P, S>
where
    S: Read,
{
    fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
        self.stream.fill_buf()
    }

    fn consume(&mut self, amt: usize) {
        self.stream.consume(amt)
    }
}

impl<P, S> Healthcheck for Session<P, S>
where
    P: Healthcheck,
{
    type Status = P::Status;

    fn get_status(&self) -> io::Result<Self::Status> {
        self.get_process().get_status()
    }

    fn is_alive(&self) -> io::Result<bool> {
        self.get_process().is_alive()
    }
}

impl<P, S> Termios for Session<P, S>
where
    P: Termios,
{
    fn is_echo(&self) -> io::Result<bool> {
        self.get_process().is_echo()
    }

    fn set_echo(&mut self, on: bool) -> io::Result<bool> {
        self.get_process_mut().set_echo(on)
    }
}

impl<P, S> NonBlocking for Session<P, S>
where
    S: NonBlocking,
{
    fn set_blocking(&mut self, on: bool) -> io::Result<()> {
        S::set_blocking(self.get_stream_mut(), on)
    }
}

#[derive(Debug)]
struct TryStream<S> {
    stream: ControlledReader<S>,
}

impl<S> TryStream<S> {
    fn into_inner(self) -> S {
        self.stream.inner.into_inner().inner
    }

    fn as_ref(&self) -> &S {
        &self.stream.inner.get_ref().inner
    }

    fn as_mut(&mut self) -> &mut S {
        &mut self.stream.inner.get_mut().inner
    }
}

impl<S> TryStream<S>
where
    S: Read,
{
    /// The function returns a new Stream from a file.
    fn new(stream: S) -> io::Result<Self> {
        Ok(Self {
            stream: ControlledReader::new(stream),
        })
    }

    fn flush_in_buffer(&mut self) {
        self.stream.flush_in_buffer();
    }
}

impl<S> TryStream<S> {
    fn keep_in_buffer(&mut self, v: &[u8]) {
        self.stream.keep_in_buffer(v);
    }

    fn get_available(&mut self) -> &[u8] {
        self.stream.get_available()
    }

    fn consume_available(&mut self, n: usize) {
        self.stream.consume_available(n)
    }
}

impl<R> TryStream<R>
where
    R: Read + NonBlocking,
{
    /// Try to read in a non-blocking mode.
    ///
    /// It raises io::ErrorKind::WouldBlock if there's nothing to read.
    fn try_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.stream.get_mut().set_blocking(false)?;

        let result = self.stream.inner.read(buf);

        // As file is DUPed changes in one descriptor affects all ones
        // so we need to make blocking file after we finished.
        self.stream.get_mut().set_blocking(true)?;

        result
    }

    #[allow(clippy::wrong_self_convention)]
    fn is_empty(&mut self) -> io::Result<bool> {
        match self.try_read(&mut []) {
            Ok(0) => Ok(true),
            Ok(_) => Ok(false),
            Err(err) if err.kind() == io::ErrorKind::WouldBlock => Ok(true),
            Err(err) => Err(err),
        }
    }

    fn read_available(&mut self) -> std::io::Result<bool> {
        self.stream.flush_in_buffer();

        let mut buf = [0; 248];
        loop {
            match self.try_read_inner(&mut buf) {
                Ok(0) => break Ok(true),
                Ok(n) => {
                    self.stream.keep_in_buffer(&buf[..n]);
                }
                Err(err) if err.kind() == io::ErrorKind::WouldBlock => break Ok(false),
                Err(err) => break Err(err),
            }
        }
    }

    fn read_available_once(&mut self, buf: &mut [u8]) -> std::io::Result<Option<usize>> {
        self.stream.flush_in_buffer();

        match self.try_read_inner(buf) {
            Ok(0) => Ok(Some(0)),
            Ok(n) => {
                self.stream.keep_in_buffer(&buf[..n]);
                Ok(Some(n))
            }
            Err(err) if err.kind() == io::ErrorKind::WouldBlock => Ok(None),
            Err(err) => Err(err),
        }
    }

    // non-buffered && non-blocking read
    fn try_read_inner(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.stream.get_mut().set_blocking(false)?;

        let result = self.stream.get_mut().read(buf);

        // As file is DUPed changes in one descriptor affects all ones
        // so we need to make blocking file after we finished.
        self.stream.get_mut().set_blocking(true)?;

        result
    }
}

impl<S> Write for TryStream<S>
where
    S: Write,
{
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.stream.inner.get_mut().inner.write(buf)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.stream.inner.get_mut().inner.flush()
    }

    fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
        self.stream.inner.get_mut().inner.write_vectored(bufs)
    }
}

impl<R> Read for TryStream<R>
where
    R: Read,
{
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.stream.inner.read(buf)
    }
}

impl<R> BufRead for TryStream<R>
where
    R: Read,
{
    fn fill_buf(&mut self) -> io::Result<&[u8]> {
        self.stream.inner.fill_buf()
    }

    fn consume(&mut self, amt: usize) {
        self.stream.inner.consume(amt)
    }
}

#[derive(Debug)]
struct ControlledReader<R> {
    inner: BufReader<BufferedReader<R>>,
}

impl<R> ControlledReader<R>
where
    R: Read,
{
    fn new(reader: R) -> Self {
        Self {
            inner: BufReader::new(BufferedReader::new(reader)),
        }
    }

    fn flush_in_buffer(&mut self) {
        // Because we have 2 buffered streams there might appear inconsistancy
        // in read operations and the data which was via `keep_in_buffer` function.
        //
        // To eliminate it we move BufReader buffer to our buffer.
        let b = self.inner.buffer().to_vec();
        self.inner.consume(b.len());
        self.keep_in_buffer(&b);
    }
}

impl<R> ControlledReader<R> {
    fn keep_in_buffer(&mut self, v: &[u8]) {
        self.inner.get_mut().buffer.extend(v);
    }

    fn get_mut(&mut self) -> &mut R {
        &mut self.inner.get_mut().inner
    }

    fn get_available(&mut self) -> &[u8] {
        &self.inner.get_ref().buffer
    }

    fn consume_available(&mut self, n: usize) {
        let _ = self.inner.get_mut().buffer.drain(..n);
    }
}

#[derive(Debug)]
struct BufferedReader<R> {
    inner: R,
    buffer: Vec<u8>,
}

impl<R> BufferedReader<R> {
    fn new(reader: R) -> Self {
        Self {
            inner: reader,
            buffer: Vec::new(),
        }
    }
}

impl<R> Read for BufferedReader<R>
where
    R: Read,
{
    fn read(&mut self, mut buf: &mut [u8]) -> std::io::Result<usize> {
        if self.buffer.is_empty() {
            self.inner.read(buf)
        } else {
            let n = buf.write(&self.buffer)?;
            let _ = self.buffer.drain(..n);
            Ok(n)
        }
    }
}

1	//! Module contains a Session structure.
2
3	use std::{
4	io::{self, BufRead, BufReader, Read, Write},
5	time::{self, Duration},
6	};
7
8	use crate::{
9	error::Error,
10	expect::Expect,
11	needle::Needle,
12	process::{Healthcheck, NonBlocking, Termios},
13	Captures,
14	};
15
16	/// Session represents a spawned process and its streams.
17	/// It controlls process and communication with it.
18	#[derive(Debug)]
19	pub struct Session<P, S> {
20	proc: P,
21	stream: TryStream<S>,
22	expect_timeout: Option<Duration>,
23	expect_lazy: bool,
24	}
25
26	impl<P, S> Session<P, S>
27	where
28	S: Read,
29	{
30	/// Creates a new session.
UNCOV 31	pub fn new(process: P, stream: S) -> io::Result<Self> {	×
UNCOV 32	let stream = TryStream::new(stream)?;	×
33
UNCOV 34	Ok(Self {	×
UNCOV 35	proc: process,	×
UNCOV 36	stream,	×
UNCOV 37	expect_timeout: Some(Duration::from_millis(10000)),	×
38	expect_lazy: false,	×
39	})
40	}
41
42	pub(crate) fn swap_stream<F, R>(mut self, new: F) -> Result<Session<P, R>, Error>
43	where
44	F: FnOnce(S) -> R,
45	R: Read,
46	{
UNCOV 47	self.stream.flush_in_buffer();	×
UNCOV 48	let buf = self.stream.get_available().to_owned();	×
49
UNCOV 50	let stream = self.stream.into_inner();	×
NEW 51	let stream = new(stream);	×
52
NEW 53	let mut session = Session::new(self.proc, stream)?;	×
UNCOV 54	session.stream.keep_in_buffer(&buf);	×
55
UNCOV 56	Ok(session)	×
57	}
58	}
59
60	impl<P, S> Session<P, S> {
61	/// Set the pty session's expect timeout.
UNCOV 62	pub fn set_expect_timeout(&mut self, expect_timeout: Option<Duration>) {	×
UNCOV 63	self.expect_timeout = expect_timeout;	×
64	}
65
66	/// Set a expect algorithm to be either gready or lazy.
67	///
68	/// Default algorithm is gready.
69	///
70	/// See [Session::expect].
UNCOV 71	pub fn set_expect_lazy(&mut self, lazy: bool) {	×
UNCOV 72	self.expect_lazy = lazy;	×
73	}
74
75	/// Get a reference to original stream.
76	pub fn get_stream(&self) -> &S {	×
77	self.stream.as_ref()	×
78	}
79
80	/// Get a mut reference to original stream.
81	pub fn get_stream_mut(&mut self) -> &mut S {	×
82	self.stream.as_mut()	×
83	}
84
85	/// Get a reference to a process running program.
UNCOV 86	pub fn get_process(&self) -> &P {	×
87	&self.proc	×
88	}
89
90	/// Get a mut reference to a process running program.
UNCOV 91	pub fn get_process_mut(&mut self) -> &mut P {	×
92	&mut self.proc	×
93	}
94	}
95
96	impl<P, S> Expect for Session<P, S>
97	where
98	S: Write + Read + NonBlocking,
99	{
100	/// Expect waits until a pattern is matched.
101	///
102	/// If the method returns [Ok] it is guaranteed that at least 1 match was found.
103	///
104	/// The match algorthm can be either
105	/// - gready
106	/// - lazy
107	///
108	/// You can set one via [Session::set_expect_lazy].
109	/// Default version is gready.
110	///
111	/// The implications are.
112	/// Imagine you use [crate::Regex] `"\d+"` to find a match.
113	/// And your process outputs `123`.
114	/// In case of lazy approach we will match `1`.
115	/// Where's in case of gready one we will match `123`.
116	///
117	/// # Example
118	///
119	#[cfg_attr(windows, doc = "```no_run")]
120	#[cfg_attr(unix, doc = "```")]
121	/// let mut p = expectrl::spawn("echo 123").unwrap();
122	/// let m = p.expect(expectrl::Regex("\\d+")).unwrap();
123	/// assert_eq!(m.get(0).unwrap(), b"123");
124	/// ```
125	///
126	#[cfg_attr(windows, doc = "```no_run")]
127	#[cfg_attr(unix, doc = "```")]
128	/// let mut p = expectrl::spawn("echo 123").unwrap();
129	/// p.set_expect_lazy(true);
130	/// let m = p.expect(expectrl::Regex("\\d+")).unwrap();
131	/// assert_eq!(m.get(0).unwrap(), b"1");
132	/// ```
133	///
134	/// This behaviour is different from [Session::check].
135	///
136	/// It returns an error if timeout is reached.
137	/// You can specify a timeout value by [Session::set_expect_timeout] method.
138	fn expect<N>(&mut self, needle: N) -> Result<Captures, Error>
139	where
140	N: Needle,
141	{
UNCOV 142	match self.expect_lazy {	×
UNCOV 143	true => self.expect_lazy(needle),	×
UNCOV 144	false => self.expect_gready(needle),	×
145	}
146	}
147
148	/// Check verifies if a pattern is matched.
149	/// Returns empty found structure if nothing found.
150	///
151	/// Is a non blocking version of [Session::expect].
152	/// But its strategy of matching is different from it.
153	/// It makes search against all bytes available.
154	///
155	/// # Example
156	///
157	#[cfg_attr(any(windows, target_os = "macos"), doc = "```no_run")]
158	#[cfg_attr(not(any(target_os = "macos", windows)), doc = "```")]
159	/// use expectrl::{spawn, Regex};
160	/// use std::time::Duration;
161	///
162	/// let mut p = spawn("echo 123").unwrap();
163	/// #
164	/// # // wait to guarantee that check echo worked out (most likely)
165	/// # std::thread::sleep(Duration::from_millis(500));
166	/// #
167	/// let m = p.check(Regex("\\d+")).unwrap();
168	/// assert_eq!(m.get(0).unwrap(), b"123");
169	/// ```
170	fn check<N>(&mut self, needle: N) -> Result<Captures, Error>
171	where
172	N: Needle,
173	{
UNCOV 174	let eof = self.stream.read_available()?;	×
UNCOV 175	let buf = self.stream.get_available();	×
176
UNCOV 177	let found = needle.check(buf, eof)?;	×
UNCOV 178	if !found.is_empty() {	×
UNCOV 179	let end_index = Captures::right_most_index(&found);	×
UNCOV 180	let involved_bytes = buf[..end_index].to_vec();	×
UNCOV 181	self.stream.consume_available(end_index);	×
UNCOV 182	return Ok(Captures::new(involved_bytes, found));	×
183	}
184
UNCOV 185	if eof {	×
186	return Err(Error::Eof);	×
187	}
188
UNCOV 189	Ok(Captures::new(Vec::new(), Vec::new()))	×
190	}
191
192	/// The functions checks if a pattern is matched.
193	/// It doesn’t consumes bytes from stream.
194	///
195	/// Its strategy of matching is different from the one in [Session::expect].
196	/// It makes search agains all bytes available.
197	///
198	/// If you want to get a matched result [Session::check] and [Session::expect] is a better option.
199	/// Because it is not guaranteed that [Session::check] or [Session::expect] with the same parameters:
200	/// - will successed even right after Session::is_matched call.
201	/// - will operate on the same bytes.
202	///
203	/// IMPORTANT:
204	///
205	/// If you call this method with [crate::Eof] pattern be aware that eof
206	/// indication MAY be lost on the next interactions.
207	/// It depends from a process you spawn.
208	/// So it might be better to use [Session::check] or [Session::expect] with Eof.
209	///
210	/// # Example
211	///
212	#[cfg_attr(windows, doc = "```no_run")]
213	#[cfg_attr(unix, doc = "```")]
214	/// use expectrl::{spawn, Regex};
215	/// use std::time::Duration;
216	///
217	/// let mut p = spawn("cat").unwrap();
218	/// p.send_line("123");
219	/// # // wait to guarantee that check echo worked out (most likely)
220	/// # std::thread::sleep(Duration::from_secs(1));
221	/// let m = p.is_matched(Regex("\\d+")).unwrap();
222	/// assert_eq!(m, true);
223	/// ```
224	fn is_matched<N>(&mut self, needle: N) -> Result<bool, Error>
225	where
226	N: Needle,
227	{
UNCOV 228	let eof = self.stream.read_available()?;	×
UNCOV 229	let buf = self.stream.get_available();	×
230
UNCOV 231	let found = needle.check(buf, eof)?;	×
UNCOV 232	if !found.is_empty() {	×
UNCOV 233	return Ok(true);	×
234	}
235
236	if eof {	×
237	return Err(Error::Eof);	×
238	}
239
240	Ok(false)
241	}
242
243	/// Send text to child’s STDIN.
244	///
245	/// You can also use methods from [std::io::Write] instead.
246	///
247	/// # Example
248	///
249	/// ```
250	/// use expectrl::{spawn, ControlCode};
251	///
252	/// let mut proc = spawn("cat").unwrap();
253	///
254	/// proc.send("Hello");
255	/// proc.send(b"World");
256	/// proc.send(ControlCode::try_from("^C").unwrap());
257	/// ```
258	fn send<B>(&mut self, buf: B) -> Result<(), Error>
259	where
260	B: AsRef<[u8]>,
261	{
NEW 262	self.stream.write_all(buf.as_ref())?;	×
263
NEW 264	Ok(())	×
265	}
266
267	/// Send a line to child’s STDIN.
268	///
269	/// # Example
270	///
271	/// ```
272	/// use expectrl::{spawn, ControlCode};
273	///
274	/// let mut proc = spawn("cat").unwrap();
275	///
276	/// proc.send_line("Hello");
277	/// proc.send_line(b"World");
278	/// proc.send_line(ControlCode::try_from("^C").unwrap());
279	/// ```
280	fn send_line<B>(&mut self, buf: B) -> Result<(), Error>
281	where
282	B: AsRef<[u8]>,
283	{
284	#[cfg(windows)]
285	const LINE_ENDING: &[u8] = b"\r\n";
286	#[cfg(not(windows))]
287	const LINE_ENDING: &[u8] = b"\n";
288
UNCOV 289	self.stream.write_all(buf.as_ref())?;	×
UNCOV 290	self.write_all(LINE_ENDING)?;	×
291
UNCOV 292	Ok(())	×
293	}
294	}
295
296	impl<P, S> Session<P, S>
297	where
298	S: Read + NonBlocking,
299	{
300	/// Expect which fills as much as possible to the buffer.
301	///
302	/// See [Session::expect].
303	fn expect_gready<N>(&mut self, needle: N) -> Result<Captures, Error>
304	where
305	N: Needle,
306	{
NEW 307	let start = time::Instant::now();	×
NEW 308	loop {	×
NEW 309	let eof = self.stream.read_available()?;	×
NEW 310	let data = self.stream.get_available();	×
311
NEW 312	let found = needle.check(data, eof)?;	×
NEW 313	if !found.is_empty() {	×
NEW 314	let end_index = Captures::right_most_index(&found);	×
NEW 315	let involved_bytes = data[..end_index].to_vec();	×
NEW 316	self.stream.consume_available(end_index);	×
317
NEW 318	return Ok(Captures::new(involved_bytes, found));	×
319	}
320
NEW 321	if eof {	×
NEW 322	return Err(Error::Eof);	×
323	}
324
NEW 325	if let Some(timeout) = self.expect_timeout {	×
NEW 326	if start.elapsed() > timeout {	×
NEW 327	return Err(Error::ExpectTimeout);	×
328	}
329	}
330	}
331	}
332
333	/// Expect which reads byte by byte.
334	///
335	/// See [Session::expect].
336	fn expect_lazy<N>(&mut self, needle: N) -> Result<Captures, Error>
337	where
338	N: Needle,
339	{
NEW 340	let mut checking_data_length = 0;	×
NEW 341	let mut eof = false;	×
NEW 342	let start = time::Instant::now();	×
NEW 343	loop {	×
NEW 344	let mut available = self.stream.get_available();	×
NEW 345	if checking_data_length == available.len() {	×
346	// We read by byte to make things as lazy as possible.
347	//
348	// It's chose is important in using Regex as a Needle.
349	// Imagine we have a `\d+` regex.
350	// Using such buffer will match string `2` imidiately eventhough right after might be other digit.
351	//
352	// The second reason is
353	// if we wouldn't read by byte EOF indication could be lost.
354	// And next blocking std::io::Read operation could be blocked forever.
355	//
356	// We could read all data available via `read_available` to reduce IO operations,
357	// but in such case we would need to keep a EOF indicator internally in stream,
358	// which is OK if EOF happens onces, but I am not sure if this is a case.
NEW 359	eof = self.stream.read_available_once(&mut [0; 1])? == Some(0);	×
NEW 360	available = self.stream.get_available();	×
361	}
362
363	// We intentinally not increase the counter
364	// and run check one more time even though the data isn't changed.
365	// Because it may be important for custom implementations of Needle.
NEW 366	if checking_data_length < available.len() {	×
NEW 367	checking_data_length += 1;	×
368	}
369
NEW 370	let data = &available[..checking_data_length];	×
371
NEW 372	let found = needle.check(data, eof)?;	×
NEW 373	if !found.is_empty() {	×
NEW 374	let end_index = Captures::right_most_index(&found);	×
NEW 375	let involved_bytes = data[..end_index].to_vec();	×
NEW 376	self.stream.consume_available(end_index);	×
NEW 377	return Ok(Captures::new(involved_bytes, found));	×
378	}
379
NEW 380	if eof {	×
NEW 381	return Err(Error::Eof);	×
382	}
383
NEW 384	if let Some(timeout) = self.expect_timeout {	×
NEW 385	if start.elapsed() > timeout {	×
NEW 386	return Err(Error::ExpectTimeout);	×
387	}
388	}
389	}
390	}
391	}
392
393	impl<P, S> Session<P, S>
394	where
395	S: Read + NonBlocking,
396	{
397	/// Try to read in a non-blocking mode.
398	///
399	/// Returns [`std::io::ErrorKind::WouldBlock`]
400	/// in case there's nothing to read.
UNCOV 401	pub fn try_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {	×
UNCOV 402	self.stream.try_read(buf)	×
403	}
404
405	/// Verifyes if stream is empty or not.
UNCOV 406	pub fn is_empty(&mut self) -> io::Result<bool> {	×
UNCOV 407	self.stream.is_empty()	×
408	}
409	}
410
411	impl<P, S> Write for Session<P, S>
412	where
413	S: Write,
414	{
UNCOV 415	fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {	×
UNCOV 416	self.stream.write(buf)	×
417	}
418
UNCOV 419	fn flush(&mut self) -> std::io::Result<()> {	×
UNCOV 420	self.stream.flush()	×
421	}
422
423	fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {	×
424	self.stream.write_vectored(bufs)	×
425	}
426	}
427
428	impl<P, S> Read for Session<P, S>
429	where
430	S: Read,
431	{
UNCOV 432	fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {	×
UNCOV 433	self.stream.read(buf)	×
434	}
435	}
436
437	impl<P, S> BufRead for Session<P, S>
438	where
439	S: Read,
440	{
UNCOV 441	fn fill_buf(&mut self) -> std::io::Result<&[u8]> {	×
UNCOV 442	self.stream.fill_buf()	×
443	}
444
UNCOV 445	fn consume(&mut self, amt: usize) {	×
UNCOV 446	self.stream.consume(amt)	×
447	}
448	}
449
450	impl<P, S> Healthcheck for Session<P, S>
451	where
452	P: Healthcheck,
453	{
454	type Status = P::Status;
455
NEW 456	fn get_status(&self) -> io::Result<Self::Status> {	×
NEW 457	self.get_process().get_status()	×
458	}
459
NEW 460	fn is_alive(&self) -> io::Result<bool> {	×
NEW 461	self.get_process().is_alive()	×
462	}
463	}
464
465	impl<P, S> Termios for Session<P, S>
466	where
467	P: Termios,
468	{
NEW 469	fn is_echo(&self) -> io::Result<bool> {	×
NEW 470	self.get_process().is_echo()	×
471	}
472
NEW 473	fn set_echo(&mut self, on: bool) -> io::Result<bool> {	×
NEW 474	self.get_process_mut().set_echo(on)	×
475	}
476	}
477
478	impl<P, S> NonBlocking for Session<P, S>
479	where
480	S: NonBlocking,
481	{
NEW 482	fn set_blocking(&mut self, on: bool) -> io::Result<()> {	×
NEW 483	S::set_blocking(self.get_stream_mut(), on)	×
484	}
485	}
486
487	#[derive(Debug)]
488	struct TryStream<S> {
489	stream: ControlledReader<S>,
490	}
491
492	impl<S> TryStream<S> {
UNCOV 493	fn into_inner(self) -> S {	×
UNCOV 494	self.stream.inner.into_inner().inner	×
495	}
496
497	fn as_ref(&self) -> &S {	×
498	&self.stream.inner.get_ref().inner	×
499	}
500
501	fn as_mut(&mut self) -> &mut S {	×
502	&mut self.stream.inner.get_mut().inner	×
503	}
504	}
505
506	impl<S> TryStream<S>
507	where
508	S: Read,
509	{
510	/// The function returns a new Stream from a file.
UNCOV 511	fn new(stream: S) -> io::Result<Self> {	×
UNCOV 512	Ok(Self {	×
UNCOV 513	stream: ControlledReader::new(stream),	×
514	})
515	}
516
UNCOV 517	fn flush_in_buffer(&mut self) {	×
UNCOV 518	self.stream.flush_in_buffer();	×
519	}
520	}
521
522	impl<S> TryStream<S> {
UNCOV 523	fn keep_in_buffer(&mut self, v: &[u8]) {	×
UNCOV 524	self.stream.keep_in_buffer(v);	×
525	}
526
UNCOV 527	fn get_available(&mut self) -> &[u8] {	×
UNCOV 528	self.stream.get_available()	×
529	}
530
UNCOV 531	fn consume_available(&mut self, n: usize) {	×
UNCOV 532	self.stream.consume_available(n)	×
533	}
534	}
535
536	impl<R> TryStream<R>
537	where
538	R: Read + NonBlocking,
539	{
540	/// Try to read in a non-blocking mode.
541	///
542	/// It raises io::ErrorKind::WouldBlock if there's nothing to read.
UNCOV 543	fn try_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {	×
NEW 544	self.stream.get_mut().set_blocking(false)?;	×
545
UNCOV 546	let result = self.stream.inner.read(buf);	×
547
548	// As file is DUPed changes in one descriptor affects all ones
549	// so we need to make blocking file after we finished.
NEW 550	self.stream.get_mut().set_blocking(true)?;	×
551
UNCOV 552	result	×
553	}
554
555	#[allow(clippy::wrong_self_convention)]
UNCOV 556	fn is_empty(&mut self) -> io::Result<bool> {	×
UNCOV 557	match self.try_read(&mut []) {	×
558	Ok(0) => Ok(true),
559	Ok(_) => Ok(false),
560	Err(err) if err.kind() == io::ErrorKind::WouldBlock => Ok(true),
561	Err(err) => Err(err),	×
562	}
563	}
564
UNCOV 565	fn read_available(&mut self) -> std::io::Result<bool> {	×
UNCOV 566	self.stream.flush_in_buffer();	×
567
UNCOV 568	let mut buf = [0; 248];	×
UNCOV 569	loop {	×
UNCOV 570	match self.try_read_inner(&mut buf) {	×
UNCOV 571	Ok(0) => break Ok(true),	×
UNCOV 572	Ok(n) => {	×
UNCOV 573	self.stream.keep_in_buffer(&buf[..n]);	×
574	}
UNCOV 575	Err(err) if err.kind() == io::ErrorKind::WouldBlock => break Ok(false),	×
576	Err(err) => break Err(err),	×
577	}
578	}
579	}
580
UNCOV 581	fn read_available_once(&mut self, buf: &mut [u8]) -> std::io::Result<Option<usize>> {	×
UNCOV 582	self.stream.flush_in_buffer();	×
583
UNCOV 584	match self.try_read_inner(buf) {	×
585	Ok(0) => Ok(Some(0)),	×
UNCOV 586	Ok(n) => {	×
UNCOV 587	self.stream.keep_in_buffer(&buf[..n]);	×
UNCOV 588	Ok(Some(n))	×
589	}
590	Err(err) if err.kind() == io::ErrorKind::WouldBlock => Ok(None),	×
591	Err(err) => Err(err),	×
592	}
593	}
594
595	// non-buffered && non-blocking read
UNCOV 596	fn try_read_inner(&mut self, buf: &mut [u8]) -> io::Result<usize> {	×
NEW 597	self.stream.get_mut().set_blocking(false)?;	×
598
UNCOV 599	let result = self.stream.get_mut().read(buf);	×
600
601	// As file is DUPed changes in one descriptor affects all ones
602	// so we need to make blocking file after we finished.
NEW 603	self.stream.get_mut().set_blocking(true)?;	×
604
UNCOV 605	result	×
606	}
607	}
608
609	impl<S> Write for TryStream<S>
610	where
611	S: Write,
612	{
UNCOV 613	fn write(&mut self, buf: &[u8]) -> io::Result<usize> {	×
UNCOV 614	self.stream.inner.get_mut().inner.write(buf)	×
615	}
616
UNCOV 617	fn flush(&mut self) -> io::Result<()> {	×
UNCOV 618	self.stream.inner.get_mut().inner.flush()	×
619	}
620
621	fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {	×
622	self.stream.inner.get_mut().inner.write_vectored(bufs)	×
623	}
624	}
625
626	impl<R> Read for TryStream<R>
627	where
628	R: Read,
629	{
UNCOV 630	fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {	×
UNCOV 631	self.stream.inner.read(buf)	×
632	}
633	}
634
635	impl<R> BufRead for TryStream<R>
636	where
637	R: Read,
638	{
UNCOV 639	fn fill_buf(&mut self) -> io::Result<&[u8]> {	×
UNCOV 640	self.stream.inner.fill_buf()	×
641	}
642
UNCOV 643	fn consume(&mut self, amt: usize) {	×
UNCOV 644	self.stream.inner.consume(amt)	×
645	}
646	}
647
648	#[derive(Debug)]
649	struct ControlledReader<R> {
650	inner: BufReader<BufferedReader<R>>,
651	}
652
653	impl<R> ControlledReader<R>
654	where
655	R: Read,
656	{
UNCOV 657	fn new(reader: R) -> Self {	×
658	Self {
UNCOV 659	inner: BufReader::new(BufferedReader::new(reader)),	×
660	}
661	}
662
UNCOV 663	fn flush_in_buffer(&mut self) {	×
664	// Because we have 2 buffered streams there might appear inconsistancy
665	// in read operations and the data which was via `keep_in_buffer` function.
666	//
667	// To eliminate it we move BufReader buffer to our buffer.
UNCOV 668	let b = self.inner.buffer().to_vec();	×
UNCOV 669	self.inner.consume(b.len());	×
UNCOV 670	self.keep_in_buffer(&b);	×
671	}
672	}
673
674	impl<R> ControlledReader<R> {
UNCOV 675	fn keep_in_buffer(&mut self, v: &[u8]) {	×
UNCOV 676	self.inner.get_mut().buffer.extend(v);	×
677	}
678
UNCOV 679	fn get_mut(&mut self) -> &mut R {	×
UNCOV 680	&mut self.inner.get_mut().inner	×
681	}
682
UNCOV 683	fn get_available(&mut self) -> &[u8] {	×
UNCOV 684	&self.inner.get_ref().buffer	×
685	}
686
UNCOV 687	fn consume_available(&mut self, n: usize) {	×
UNCOV 688	let _ = self.inner.get_mut().buffer.drain(..n);	×
689	}
690	}
691
692	#[derive(Debug)]
693	struct BufferedReader<R> {
694	inner: R,
695	buffer: Vec<u8>,
696	}
697
698	impl<R> BufferedReader<R> {
UNCOV 699	fn new(reader: R) -> Self {	×
700	Self {
701	inner: reader,
UNCOV 702	buffer: Vec::new(),	×
703	}
704	}
705	}
706
707	impl<R> Read for BufferedReader<R>
708	where
709	R: Read,
710	{
UNCOV 711	fn read(&mut self, mut buf: &mut [u8]) -> std::io::Result<usize> {	×
UNCOV 712	if self.buffer.is_empty() {	×
UNCOV 713	self.inner.read(buf)	×
714	} else {
UNCOV 715	let n = buf.write(&self.buffer)?;	×
UNCOV 716	let _ = self.buffer.drain(..n);	×
UNCOV 717	Ok(n)	×
718	}
719	}
720	}

zhiburt / expectrl / 8129528574

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous