8119237697

Committed 02 Mar 2024 01:11AM UTC coverage: 56.739% (+0.2%) from 56.575%

Build # 8119237697

Build Type

Pull #68

github

Committed by

web-flow

Commit Message

Merge d1e39b6e1 into e8b43ff1b

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

Run Details

226 of 300 new or added lines in 12 files covered. (75.33%)

141 existing lines in 5 files now uncovered.

1486 of 2619 relevant lines covered (56.74%)

3.46 hits per line

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

89.32

/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, Process, 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, Process, 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.
31	pub fn new(process: P, stream: S) -> io::Result<Self> {	10✔
32	let stream = TryStream::new(stream)?;	20✔
33
34	Ok(Self {	10✔
35	proc: process,	10✔
36	stream,	10✔
37	expect_timeout: Some(Duration::from_millis(10000)),	20✔
38	expect_lazy: false,	×
39	})
40	}
41
42	pub(crate) fn swap_stream<F, R>(mut self, new: F) -> Result<Session<P, R>, Error>	2✔
43	where
44	F: FnOnce(S) -> R,
45	R: Read,
46	{
47	self.stream.flush_in_buffer();	2✔
48	let buf = self.stream.get_available().to_owned();	2✔
49
50	let stream = self.stream.into_inner();	4✔
51	let stream = new(stream);	2✔
52
53	let mut session = Session::new(self.proc, stream)?;	2✔
54	session.stream.keep_in_buffer(&buf);	4✔
55
56	Ok(session)	2✔
57	}
58	}
59
60	impl<P, S> Session<P, S> {
61	/// Set the pty session's expect timeout.
62	pub fn set_expect_timeout(&mut self, expect_timeout: Option<Duration>) {	1✔
63	self.expect_timeout = expect_timeout;	1✔
64	}
65
66	/// Set a expect algorithm to be either gready or lazy.
67	///
68	/// Default algorithm is gready.
69	///
70	/// See [Session::expect].
71	pub fn set_expect_lazy(&mut self, lazy: bool) {	1✔
72	self.expect_lazy = lazy;	1✔
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 {	1✔
82	self.stream.as_mut()	1✔
83	}
84
85	/// Get a reference to a process running program.
86	pub fn get_process(&self) -> &P {	5✔
87	&self.proc	1✔
88	}
89
90	/// Get a mut reference to a process running program.
91	pub fn get_process_mut(&mut self) -> &mut P {	6✔
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>	9✔
139	where
140	N: Needle,
141	{
142	match self.expect_lazy {	9✔
143	true => self.expect_lazy(needle),	1✔
144	false => self.expect_gready(needle),	9✔
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>	6✔
171	where
172	N: Needle,
173	{
174	let eof = self.stream.read_available()?;	12✔
175	let buf = self.stream.get_available();	12✔
176
177	let found = needle.check(buf, eof)?;	6✔
178	if !found.is_empty() {	12✔
179	let end_index = Captures::right_most_index(&found);	12✔
180	let involved_bytes = buf[..end_index].to_vec();	6✔
181	self.stream.consume_available(end_index);	6✔
182	return Ok(Captures::new(involved_bytes, found));	6✔
183	}
184
185	if eof {	3✔
186	return Err(Error::Eof);	×
187	}
188
189	Ok(Captures::new(Vec::new(), Vec::new()))	6✔
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>	4✔
225	where
226	N: Needle,
227	{
228	let eof = self.stream.read_available()?;	8✔
229	let buf = self.stream.get_available();	8✔
230
231	let found = needle.check(buf, eof)?;	4✔
232	if !found.is_empty() {	8✔
233	return Ok(true);	4✔
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>	5✔
259	where
260	B: AsRef<[u8]>,
261	{
262	self.stream.write_all(buf.as_ref())?;	10✔
263
264	Ok(())	5✔
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>	9✔
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
289	self.stream.write_all(buf.as_ref())?;	18✔
290	self.write_all(LINE_ENDING)?;	27✔
291
292	Ok(())	9✔
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>	9✔
304	where
305	N: Needle,
306	{
307	let start = time::Instant::now();	18✔
308	loop {	2✔
309	let eof = self.stream.read_available()?;	9✔
310	let data = self.stream.get_available();	18✔
311
312	let found = needle.check(data, eof)?;	9✔
313	if !found.is_empty() {	18✔
314	let end_index = Captures::right_most_index(&found);	16✔
315	let involved_bytes = data[..end_index].to_vec();	8✔
316	self.stream.consume_available(end_index);	8✔
317
318	return Ok(Captures::new(involved_bytes, found));	8✔
319	}
320
321	if eof {	3✔
322	return Err(Error::Eof);	1✔
323	}
324
325	if let Some(timeout) = self.expect_timeout {	4✔
326	if start.elapsed() > timeout {	4✔
327	return Err(Error::ExpectTimeout);	1✔
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>	1✔
337	where
338	N: Needle,
339	{
340	let mut checking_data_length = 0;	1✔
341	let mut eof = false;	1✔
342	let start = time::Instant::now();	2✔
343	loop {	1✔
344	let mut available = self.stream.get_available();	1✔
345	if checking_data_length == available.len() {	2✔
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.
359	eof = self.stream.read_available_once(&mut [0; 1])? == Some(0);	6✔
360	available = self.stream.get_available();	1✔
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.
366	if checking_data_length < available.len() {	2✔
367	checking_data_length += 1;	1✔
368	}
369
370	let data = &available[..checking_data_length];	2✔
371
372	let found = needle.check(data, eof)?;	1✔
373	if !found.is_empty() {	2✔
374	let end_index = Captures::right_most_index(&found);	2✔
375	let involved_bytes = data[..end_index].to_vec();	1✔
376	self.stream.consume_available(end_index);	1✔
377	return Ok(Captures::new(involved_bytes, found));	1✔
378	}
379
380	if eof {	1✔
NEW 381	return Err(Error::Eof);	×
382	}
383
384	if let Some(timeout) = self.expect_timeout {	2✔
385	if start.elapsed() > timeout {	2✔
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.
401	pub fn try_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {	1✔
402	self.stream.try_read(buf)	1✔
403	}
404
405	/// Verifyes if stream is empty or not.
406	pub fn is_empty(&mut self) -> io::Result<bool> {	1✔
407	self.stream.is_empty()	1✔
408	}
409	}
410
411	impl<P, S> Write for Session<P, S>
412	where
413	S: Write,
414	{
415	fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {	9✔
416	self.stream.write(buf)	9✔
417	}
418
419	fn flush(&mut self) -> std::io::Result<()> {	1✔
420	self.stream.flush()	1✔
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	{
432	fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {	7✔
433	self.stream.read(buf)	7✔
434	}
435	}
436
437	impl<P, S> BufRead for Session<P, S>
438	where
439	S: Read,
440	{
441	fn fill_buf(&mut self) -> std::io::Result<&[u8]> {	4✔
442	self.stream.fill_buf()	4✔
443	}
444
445	fn consume(&mut self, amt: usize) {	4✔
446	self.stream.consume(amt)	4✔
447	}
448	}
449
450	impl<P, S> Healthcheck for Session<P, S>
451	where
452	P: Healthcheck,
453	{
454	type Status = P::Status;
455
456	fn get_status(&self) -> io::Result<Self::Status> {	2✔
457	self.get_process().get_status()	2✔
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	{
469	fn is_echo(&self) -> io::Result<bool> {	2✔
470	self.get_process().is_echo()	2✔
471	}
472
473	fn set_echo(&mut self, on: bool) -> io::Result<bool> {	2✔
474	self.get_process_mut().set_echo(on)	2✔
475	}
476	}
477
478	impl<P, S> NonBlocking for Session<P, S>
479	where
480	S: NonBlocking,
481	{
482	fn set_blocking(&mut self, on: bool) -> io::Result<()> {	1✔
483	S::set_blocking(self.get_stream_mut(), on)	1✔
484	}
485	}
486
487	#[derive(Debug)]
488	struct TryStream<S> {
489	stream: ControlledReader<S>,
490	}
491
492	impl<S> TryStream<S> {
493	fn into_inner(self) -> S {	2✔
494	self.stream.inner.into_inner().inner	2✔
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 {	1✔
502	&mut self.stream.inner.get_mut().inner	1✔
503	}
504	}
505
506	impl<S> TryStream<S>
507	where
508	S: Read,
509	{
510	/// The function returns a new Stream from a file.
511	fn new(stream: S) -> io::Result<Self> {	10✔
512	Ok(Self {	10✔
513	stream: ControlledReader::new(stream),	10✔
514	})
515	}
516
517	fn flush_in_buffer(&mut self) {	2✔
518	self.stream.flush_in_buffer();	2✔
519	}
520	}
521
522	impl<S> TryStream<S> {
523	fn keep_in_buffer(&mut self, v: &[u8]) {	2✔
524	self.stream.keep_in_buffer(v);	2✔
525	}
526
527	fn get_available(&mut self) -> &[u8] {	7✔
528	self.stream.get_available()	7✔
529	}
530
531	fn consume_available(&mut self, n: usize) {	6✔
532	self.stream.consume_available(n)	6✔
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.
543	fn try_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {	1✔
544	self.stream.get_mut().set_blocking(false)?;	1✔
545
546	let result = self.stream.inner.read(buf);	1✔
547
548	// As file is DUPed changes in one descriptor affects all ones
549	// so we need to make blocking file after we finished.
550	self.stream.get_mut().set_blocking(true)?;	3✔
551
552	result	1✔
553	}
554
555	#[allow(clippy::wrong_self_convention)]
556	fn is_empty(&mut self) -> io::Result<bool> {	1✔
557	match self.try_read(&mut []) {	1✔
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
565	fn read_available(&mut self) -> std::io::Result<bool> {	6✔
566	self.stream.flush_in_buffer();	6✔
567
568	let mut buf = [0; 248];	6✔
569	loop {	6✔
570	match self.try_read_inner(&mut buf) {	6✔
571	Ok(0) => break Ok(true),	3✔
572	Ok(n) => {	6✔
573	self.stream.keep_in_buffer(&buf[..n]);	12✔
574	}
575	Err(err) if err.kind() == io::ErrorKind::WouldBlock => break Ok(false),	18✔
576	Err(err) => break Err(err),	×
577	}
578	}
579	}
580
581	fn read_available_once(&mut self, buf: &mut [u8]) -> std::io::Result<Option<usize>> {	1✔
582	self.stream.flush_in_buffer();	1✔
583
584	match self.try_read_inner(buf) {	1✔
585	Ok(0) => Ok(Some(0)),	×
586	Ok(n) => {	1✔
587	self.stream.keep_in_buffer(&buf[..n]);	2✔
588	Ok(Some(n))	1✔
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
596	fn try_read_inner(&mut self, buf: &mut [u8]) -> io::Result<usize> {	6✔
597	self.stream.get_mut().set_blocking(false)?;	6✔
598
599	let result = self.stream.get_mut().read(buf);	6✔
600
601	// As file is DUPed changes in one descriptor affects all ones
602	// so we need to make blocking file after we finished.
603	self.stream.get_mut().set_blocking(true)?;	18✔
604
605	result	6✔
606	}
607	}
608
609	impl<S> Write for TryStream<S>
610	where
611	S: Write,
612	{
613	fn write(&mut self, buf: &[u8]) -> io::Result<usize> {	9✔
614	self.stream.inner.get_mut().inner.write(buf)	9✔
615	}
616
617	fn flush(&mut self) -> io::Result<()> {	1✔
618	self.stream.inner.get_mut().inner.flush()	1✔
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	{
630	fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {	7✔
631	self.stream.inner.read(buf)	7✔
632	}
633	}
634
635	impl<R> BufRead for TryStream<R>
636	where
637	R: Read,
638	{
639	fn fill_buf(&mut self) -> io::Result<&[u8]> {	4✔
640	self.stream.inner.fill_buf()	4✔
641	}
642
643	fn consume(&mut self, amt: usize) {	4✔
644	self.stream.inner.consume(amt)	4✔
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	{
657	fn new(reader: R) -> Self {	10✔
658	Self {
659	inner: BufReader::new(BufferedReader::new(reader)),	10✔
660	}
661	}
662
663	fn flush_in_buffer(&mut self) {	7✔
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.
668	let b = self.inner.buffer().to_vec();	7✔
669	self.inner.consume(b.len());	14✔
670	self.keep_in_buffer(&b);	7✔
671	}
672	}
673
674	impl<R> ControlledReader<R> {
675	fn keep_in_buffer(&mut self, v: &[u8]) {	9✔
676	self.inner.get_mut().buffer.extend(v);	9✔
677	}
678
679	fn get_mut(&mut self) -> &mut R {	7✔
680	&mut self.inner.get_mut().inner	7✔
681	}
682
683	fn get_available(&mut self) -> &[u8] {	7✔
684	&self.inner.get_ref().buffer	7✔
685	}
686
687	fn consume_available(&mut self, n: usize) {	6✔
688	let _ = self.inner.get_mut().buffer.drain(..n);	6✔
689	}
690	}
691
692	#[derive(Debug)]
693	struct BufferedReader<R> {
694	inner: R,
695	buffer: Vec<u8>,
696	}
697
698	impl<R> BufferedReader<R> {
699	fn new(reader: R) -> Self {	10✔
700	Self {
701	inner: reader,
702	buffer: Vec::new(),	10✔
703	}
704	}
705	}
706
707	impl<R> Read for BufferedReader<R>
708	where
709	R: Read,
710	{
711	fn read(&mut self, mut buf: &mut [u8]) -> std::io::Result<usize> {	9✔
712	if self.buffer.is_empty() {	12✔
713	self.inner.read(buf)	6✔
714	} else {
715	let n = buf.write(&self.buffer)?;	6✔
716	let _ = self.buffer.drain(..n);	3✔
717	Ok(n)	3✔
718	}
719	}
720	}

zhiburt / expectrl / 8119237697

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