29192566528

Committed 12 Jul 2026 12:16PM UTC coverage: 89.462% (+29.2%) from 60.289%

Build # 29192566528

Build Type

push

github

Committed by

lpenz

Commit Message

fix: make console-subscriber init panic-safe for test compatibility

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

10 existing lines in 1 file now uncovered.

832 of 930 relevant lines covered (89.46%)

4.08 hits per line

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

93.29

/src/process_wrapper.rs

// Copyright (C) 2025 Leandro Lisboa Penz <lpenz@lpenz.org>
// This file is subject to the terms and conditions defined in
// file 'LICENSE', which is part of this source code package.

//! Wrapper for process functions.
//!
//! # `Cmd`
//!
//! The [`Cmd`] type has an inner `Vec<String>` that we can turn into
//! a [`tokio::process::Command`]. It implements `Clone`, which we use
//! to spawn the same process multiple times.
//!
//! # `ProcessStream`
//!
//! The [`ProcessStream`] type wraps [`tokio_process_stream`] in order
//! to provide an [`Item`] that implements `Eq` which we can then use
//! for testing.

use color_eyre::Result;
use nix::sys::signal::Signal;
use std::collections::VecDeque;
use std::fmt;
use std::io;
use std::os::unix::process::ExitStatusExt;
use std::pin::Pin;
use std::process::ExitStatus;
use std::process::Stdio;
use std::task::{Context, Poll};
use tokio::process::Child;
use tokio::process::Command;
use tokio_process_stream as tps;
use tokio_stream::Stream;
use tracing::instrument;

// Command wrapper ///////////////////////////////////////////////////

/// A [`tokio::process::Command`] pseudo-wrapper that `impl Clone`.
#[derive(Debug, Default, Clone)]
pub struct Cmd(Vec<String>);

impl From<&Cmd> for Command {
    fn from(cmd: &Cmd) -> Command {
        let mut command = Command::new(&cmd.0[0]);
        command.args(cmd.0.iter().skip(1));
        command.stdin(Stdio::null());
        command.stdout(Stdio::piped());
        command.stderr(Stdio::piped());
        command
    }
}

impl From<Vec<String>> for Cmd {
    fn from(s: Vec<String>) -> Cmd {
        Self(s)
    }
}

impl From<&[&str]> for Cmd {
    fn from(s: &[&str]) -> Cmd {
        Self(s.iter().map(|s| s.to_string()).collect::<Vec<_>>())
    }
}

impl fmt::Display for Cmd {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let joined = self.0.join(" ");
        write!(f, "{joined}")
    }
}

// Exit status ///////////////////////////////////////////////////////

/// A [`std::process::ExitStatus`] pseudo-wrapper that impl Clone
/// and custom Display.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub enum ExitSts {
    #[default]
    Success,
    Code(u8),
    Signal(i32),
}

impl ExitSts {
    pub fn success(&self) -> bool {
        self == &ExitSts::Success
    }
}

impl From<ExitStatus> for ExitSts {
    fn from(sts: ExitStatus) -> ExitSts {
        if sts.success() {
            ExitSts::Success
        } else if let Some(code) = sts.code() {
            debug_assert!(
                (0..=255).contains(&code),
                "exit code {code} out of u8 range"
            );
            ExitSts::Code(code as u8)
        } else if let Some(signal) = sts.signal() {
            ExitSts::Signal(signal)
        } else {
            panic!("Unable to figure out exit status {sts:?}")
        }
    }
}

impl fmt::Display for ExitSts {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ExitSts::Success => write!(f, "success"),
            ExitSts::Code(code) => write!(f, "code {code}"),
            ExitSts::Signal(signal) => {
                if let Ok(s) = Signal::try_from(*signal) {
                    write!(f, "signal {signal} ({s})")
                } else {
                    write!(f, "signal {signal}")
                }
            }
        }
    }
}

// ProcessStream /////////////////////////////////////////////////////

/// A clonable, Eq replacement for [`tokio_process_stream::Item`]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Item {
    /// A stdout line printed by the process.
    Stdout(String),
    /// A stderr line printed by the process.
    Stderr(String),
    /// The [`ExitSts`], yielded after the process exits.
    Done(Result<ExitSts, io::ErrorKind>),
}

impl From<tps::Item<String>> for Item {
    fn from(item: tps::Item<String>) -> Self {
        match item {
            tps::Item::Stdout(s) => Item::Stdout(s),
            tps::Item::Stderr(s) => Item::Stderr(s),
            tps::Item::Done(result) => Item::Done(match result {
                Ok(sts) => Ok(sts.into()),
                Err(err) => Err(err.kind()),
            }),
        }
    }
}

/// A wrapper for [`tokio_process_stream::ProcessLineStream`].
///
/// Also provides a virtual implementation for use in tests.
pub enum ProcessStream {
    /// Wrapper for [`tokio_process_stream::ProcessLineStream`].
    Real { stream: Box<tps::ProcessLineStream> },
    /// Mock for a running process stream that just returns items from
    /// a list. Useful for testing.
    Virtual { items: VecDeque<Item> },
}

impl ProcessStream {
    /// Return a mutable reference to the child object
    pub fn child_mut(&mut self) -> Option<&mut Child> {
        if let ProcessStream::Real { stream } = self {
            stream.child_mut()
        } else {
            None
        }
    }
}

impl std::fmt::Debug for ProcessStream {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ProcessStream::Real { stream: _ } => f.debug_struct("ProcessStream::Real"),
            ProcessStream::Virtual { items: _ } => f.debug_struct("ProcessStream::Virtual"),
        }
        .finish()
    }
}

impl From<tps::ProcessLineStream> for ProcessStream {
    fn from(stream: tps::ProcessLineStream) -> Self {
        ProcessStream::Real {
            stream: Box::new(stream),
        }
    }
}

impl From<VecDeque<Item>> for ProcessStream {
    fn from(items: VecDeque<Item>) -> Self {
        ProcessStream::Virtual { items }
    }
}

impl Stream for ProcessStream {
    type Item = Item;

    #[instrument(level = "debug", ret, skip(cx))]
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        match this {
            ProcessStream::Real { stream } => {
                let next = Pin::new(stream).poll_next(cx);
                match next {
                    Poll::Ready(opt) => Poll::Ready(opt.map(|i| i.into())),
                    Poll::Pending => Poll::Pending,
                }
            }
            ProcessStream::Virtual { items } => Poll::Ready(items.pop_front()),
        }
    }
}

// Tests /////////////////////////////////////////////////////////////

#[cfg(test)]
pub mod test {
    use color_eyre::Result;
    use color_eyre::eyre::eyre;
    use tokio_stream::StreamExt;

    use super::*;

    async fn stream_cmd(cmdstr: &[&str]) -> Result<ProcessStream> {
        let cmd = Cmd::from(cmdstr);
        let process_stream = tps::ProcessLineStream::try_from(Command::from(&cmd))?;
        Ok(ProcessStream::from(process_stream))
    }

    async fn stream_next<T>(stream: &mut T) -> Result<Item>
    where
        T: StreamExt<Item = Item> + std::marker::Unpin + Send + 'static,
    {
        stream.next().await.ok_or(eyre!("no item received"))
    }

    async fn assert_closed<T>(stream: &mut T)
    where
        T: StreamExt<Item = Item> + std::marker::Unpin + Send + 'static,
    {
        assert_eq!(stream.next().await, None);
    }

    #[tokio::test]
    async fn test_true() -> Result<()> {
        let mut stream = stream_cmd(&["true"]).await?;
        let item = stream_next(&mut stream).await?;
        let Item::Done(sts) = item else {
            return Err(eyre!("unexpected stream item {:?}", item));
        };
        assert!(sts.unwrap().success());
        assert!(stream.next().await.is_none());
        assert_closed(&mut stream).await;
        Ok(())
    }

    #[tokio::test]
    async fn test_false() -> Result<()> {
        let mut stream = stream_cmd(&["false"]).await?;
        let item = stream_next(&mut stream).await?;
        let Item::Done(sts) = item else {
            return Err(eyre!("unexpected stream item {:?}", item));
        };
        assert!(!sts.unwrap().success());
        assert_closed(&mut stream).await;
        Ok(())
    }

    #[tokio::test]
    async fn test_echo() -> Result<()> {
        let mut stream = stream_cmd(&["echo", "test"]).await?;
        let item = stream_next(&mut stream).await?;
        let Item::Stdout(s) = item else {
            return Err(eyre!("unexpected stream item {:?}", item));
        };
        assert_eq!(s, "test");
        let item = stream_next(&mut stream).await?;
        let Item::Done(sts) = item else {
            return Err(eyre!("unexpected stream item {:?}", item));
        };
        assert!(sts.unwrap().success());
        assert_closed(&mut stream).await;
        Ok(())
    }

    #[tokio::test]
    async fn test_stderr() -> Result<()> {
        let mut stream = stream_cmd(&["/bin/sh", "-c", "echo test >&2"]).await?;
        let item = stream_next(&mut stream).await?;
        let Item::Stderr(s) = item else {
            return Err(eyre!("unexpected stream item {:?}", item));
        };
        assert_eq!(s, "test");
        let item = stream_next(&mut stream).await?;
        let Item::Done(sts) = item else {
            return Err(eyre!("unexpected stream item {:?}", item));
        };
        assert!(sts.unwrap().success());
        assert_closed(&mut stream).await;
        Ok(())
    }

    #[tokio::test]
    async fn test_kill() -> Result<()> {
        let mut stream = stream_cmd(&["/bin/sh", "-c", "sleep 60"]).await?;
        if let Some(child) = stream.child_mut() {
            let _ = child.start_kill();
        }
        let item = stream_next(&mut stream).await?;
        assert_eq!(item, Item::Done(Ok(ExitSts::Signal(9))));
        assert_closed(&mut stream).await;
        Ok(())
    }

    #[test]
    fn test_exitsts_display() {
        assert_eq!(format!("{}", ExitSts::Success), "success");
        assert_eq!(format!("{}", ExitSts::Code(42)), "code 42");
        assert_eq!(format!("{}", ExitSts::Signal(1)), "signal 1 (SIGHUP)");
        assert_eq!(format!("{}", ExitSts::Signal(12345)), "signal 12345");
    }
}

1	// Copyright (C) 2025 Leandro Lisboa Penz <lpenz@lpenz.org>
2	// This file is subject to the terms and conditions defined in
3	// file 'LICENSE', which is part of this source code package.
4
5	//! Wrapper for process functions.
6	//!
7	//! # `Cmd`
8	//!
9	//! The [`Cmd`] type has an inner `Vec<String>` that we can turn into
10	//! a [`tokio::process::Command`]. It implements `Clone`, which we use
11	//! to spawn the same process multiple times.
12	//!
13	//! # `ProcessStream`
14	//!
15	//! The [`ProcessStream`] type wraps [`tokio_process_stream`] in order
16	//! to provide an [`Item`] that implements `Eq` which we can then use
17	//! for testing.
18
19	use color_eyre::Result;
20	use nix::sys::signal::Signal;
21	use std::collections::VecDeque;
22	use std::fmt;
23	use std::io;
24	use std::os::unix::process::ExitStatusExt;
25	use std::pin::Pin;
26	use std::process::ExitStatus;
27	use std::process::Stdio;
28	use std::task::{Context, Poll};
29	use tokio::process::Child;
30	use tokio::process::Command;
31	use tokio_process_stream as tps;
32	use tokio_stream::Stream;
33	use tracing::instrument;
34
35	// Command wrapper ///////////////////////////////////////////////////
36
37	/// A [`tokio::process::Command`] pseudo-wrapper that `impl Clone`.
38	#[derive(Debug, Default, Clone)]
39	pub struct Cmd(Vec<String>);
40
41	impl From<&Cmd> for Command {
42	fn from(cmd: &Cmd) -> Command {	11✔
43	let mut command = Command::new(&cmd.0[0]);	11✔
44	command.args(cmd.0.iter().skip(1));	11✔
45	command.stdin(Stdio::null());	11✔
46	command.stdout(Stdio::piped());	11✔
47	command.stderr(Stdio::piped());	11✔
48	command	11✔
49	}	11✔
50	}
51
52	impl From<Vec<String>> for Cmd {
53	fn from(s: Vec<String>) -> Cmd {	5✔
54	Self(s)	5✔
55	}	5✔
56	}
57
58	impl From<&[&str]> for Cmd {
59	fn from(s: &[&str]) -> Cmd {	6✔
60	Self(s.iter().map(\|s\| s.to_string()).collect::<Vec<_>>())	11✔
61	}	6✔
62	}
63
64	impl fmt::Display for Cmd {
65	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {	4✔
66	let joined = self.0.join(" ");	4✔
67	write!(f, "{joined}")	4✔
68	}	4✔
69	}
70
71	// Exit status ///////////////////////////////////////////////////////
72
73	/// A [`std::process::ExitStatus`] pseudo-wrapper that impl Clone
74	/// and custom Display.
75	#[derive(Debug, Default, Clone, PartialEq, Eq)]
76	pub enum ExitSts {
77	#[default]
78	Success,
79	Code(u8),
80	Signal(i32),
81	}
82
83	impl ExitSts {
84	pub fn success(&self) -> bool {	10✔
85	self == &ExitSts::Success	10✔
86	}	10✔
87	}
88
89	impl From<ExitStatus> for ExitSts {
90	fn from(sts: ExitStatus) -> ExitSts {	9✔
91	if sts.success() {	9✔
92	ExitSts::Success	6✔
93	} else if let Some(code) = sts.code() {	3✔
94	debug_assert!(	2✔
95	(0..=255).contains(&code),	2✔
96	"exit code {code} out of u8 range"
97	);
98	ExitSts::Code(code as u8)	2✔
99	} else if let Some(signal) = sts.signal() {	1✔
100	ExitSts::Signal(signal)	1✔
101	} else {
UNCOV 102	panic!("Unable to figure out exit status {sts:?}")	×
103	}
104	}	9✔
105	}
106
107	impl fmt::Display for ExitSts {
108	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {	7✔
109	match self {	7✔
110	ExitSts::Success => write!(f, "success"),	3✔
111	ExitSts::Code(code) => write!(f, "code {code}"),	2✔
112	ExitSts::Signal(signal) => {	2✔
113	if let Ok(s) = Signal::try_from(*signal) {	2✔
114	write!(f, "signal {signal} ({s})")	1✔
115	} else {
116	write!(f, "signal {signal}")	1✔
117	}
118	}
119	}
120	}	7✔
121	}
122
123	// ProcessStream /////////////////////////////////////////////////////
124
125	/// A clonable, Eq replacement for [`tokio_process_stream::Item`]
126	#[derive(Debug, Clone, PartialEq, Eq)]
127	pub enum Item {
128	/// A stdout line printed by the process.
129	Stdout(String),
130	/// A stderr line printed by the process.
131	Stderr(String),
132	/// The [`ExitSts`], yielded after the process exits.
133	Done(Result<ExitSts, io::ErrorKind>),
134	}
135
136	impl From<tps::Item<String>> for Item {
137	fn from(item: tps::Item<String>) -> Self {	12✔
138	match item {	12✔
139	tps::Item::Stdout(s) => Item::Stdout(s),	2✔
140	tps::Item::Stderr(s) => Item::Stderr(s),	1✔
141	tps::Item::Done(result) => Item::Done(match result {	9✔
142	Ok(sts) => Ok(sts.into()),	9✔
UNCOV 143	Err(err) => Err(err.kind()),	×
144	}),
145	}
146	}	12✔
147	}
148
149	/// A wrapper for [`tokio_process_stream::ProcessLineStream`].
150	///
151	/// Also provides a virtual implementation for use in tests.
152	pub enum ProcessStream {
153	/// Wrapper for [`tokio_process_stream::ProcessLineStream`].
154	Real { stream: Box<tps::ProcessLineStream> },
155	/// Mock for a running process stream that just returns items from
156	/// a list. Useful for testing.
157	Virtual { items: VecDeque<Item> },
158	}
159
160	impl ProcessStream {
161	/// Return a mutable reference to the child object
162	pub fn child_mut(&mut self) -> Option<&mut Child> {	1✔
163	if let ProcessStream::Real { stream } = self {	1✔
164	stream.child_mut()	1✔
165	} else {
UNCOV 166	None	×
167	}
168	}	1✔
169	}
170
171	impl std::fmt::Debug for ProcessStream {
172	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	1✔
173	match self {	1✔
UNCOV 174	ProcessStream::Real { stream: _ } => f.debug_struct("ProcessStream::Real"),	×
175	ProcessStream::Virtual { items: _ } => f.debug_struct("ProcessStream::Virtual"),	1✔
176	}
177	.finish()	1✔
178	}	1✔
179	}
180
181	impl From<tps::ProcessLineStream> for ProcessStream {
182	fn from(stream: tps::ProcessLineStream) -> Self {	9✔
183	ProcessStream::Real {	9✔
184	stream: Box::new(stream),	9✔
185	}	9✔
186	}	9✔
187	}
188
189	impl From<VecDeque<Item>> for ProcessStream {
190	fn from(items: VecDeque<Item>) -> Self {	5✔
191	ProcessStream::Virtual { items }	5✔
192	}	5✔
193	}
194
195	impl Stream for ProcessStream {
196	type Item = Item;
197
198	#[instrument(level = "debug", ret, skip(cx))]
199	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {	45✔
200	let this = self.get_mut();
201	match this {
202	ProcessStream::Real { stream } => {
203	let next = Pin::new(stream).poll_next(cx);
204	match next {
205	Poll::Ready(opt) => Poll::Ready(opt.map(\|i\| i.into())),	12✔
206	Poll::Pending => Poll::Pending,
207	}
208	}
209	ProcessStream::Virtual { items } => Poll::Ready(items.pop_front()),
210	}
211	}	45✔
212	}
213
214	// Tests /////////////////////////////////////////////////////////////
215
216	#[cfg(test)]
217	pub mod test {
218	use color_eyre::Result;
219	use color_eyre::eyre::eyre;
220	use tokio_stream::StreamExt;
221
222	use super::*;
223
224	async fn stream_cmd(cmdstr: &[&str]) -> Result<ProcessStream> {	5✔
225	let cmd = Cmd::from(cmdstr);	5✔
226	let process_stream = tps::ProcessLineStream::try_from(Command::from(&cmd))?;	5✔
227	Ok(ProcessStream::from(process_stream))	5✔
228	}	5✔
229
230	async fn stream_next<T>(stream: &mut T) -> Result<Item>	7✔
231	where	7✔
232	T: StreamExt<Item = Item> + std::marker::Unpin + Send + 'static,	7✔
233	{	7✔
234	stream.next().await.ok_or(eyre!("no item received"))	7✔
235	}	7✔
236
237	async fn assert_closed<T>(stream: &mut T)	5✔
238	where	5✔
239	T: StreamExt<Item = Item> + std::marker::Unpin + Send + 'static,	5✔
240	{	5✔
241	assert_eq!(stream.next().await, None);	5✔
242	}	5✔
243
244	#[tokio::test]
245	async fn test_true() -> Result<()> {	1✔
246	let mut stream = stream_cmd(&["true"]).await?;	1✔
247	let item = stream_next(&mut stream).await?;	1✔
248	let Item::Done(sts) = item else {	1✔
UNCOV 249	return Err(eyre!("unexpected stream item {:?}", item));	×
250	};
251	assert!(sts.unwrap().success());	1✔
252	assert!(stream.next().await.is_none());	1✔
253	assert_closed(&mut stream).await;	1✔
254	Ok(())	2✔
255	}	1✔
256
257	#[tokio::test]
258	async fn test_false() -> Result<()> {	1✔
259	let mut stream = stream_cmd(&["false"]).await?;	1✔
260	let item = stream_next(&mut stream).await?;	1✔
261	let Item::Done(sts) = item else {	1✔
UNCOV 262	return Err(eyre!("unexpected stream item {:?}", item));	×
263	};
264	assert!(!sts.unwrap().success());	1✔
265	assert_closed(&mut stream).await;	1✔
266	Ok(())	2✔
267	}	1✔
268
269	#[tokio::test]
270	async fn test_echo() -> Result<()> {	1✔
271	let mut stream = stream_cmd(&["echo", "test"]).await?;	1✔
272	let item = stream_next(&mut stream).await?;	1✔
273	let Item::Stdout(s) = item else {	1✔
UNCOV 274	return Err(eyre!("unexpected stream item {:?}", item));	×
275	};
276	assert_eq!(s, "test");	1✔
277	let item = stream_next(&mut stream).await?;	1✔
278	let Item::Done(sts) = item else {	1✔
UNCOV 279	return Err(eyre!("unexpected stream item {:?}", item));	×
280	};
281	assert!(sts.unwrap().success());	1✔
282	assert_closed(&mut stream).await;	1✔
283	Ok(())	2✔
284	}	1✔
285
286	#[tokio::test]
287	async fn test_stderr() -> Result<()> {	1✔
288	let mut stream = stream_cmd(&["/bin/sh", "-c", "echo test >&2"]).await?;	1✔
289	let item = stream_next(&mut stream).await?;	1✔
290	let Item::Stderr(s) = item else {	1✔
UNCOV 291	return Err(eyre!("unexpected stream item {:?}", item));	×
292	};
293	assert_eq!(s, "test");	1✔
294	let item = stream_next(&mut stream).await?;	1✔
295	let Item::Done(sts) = item else {	1✔
UNCOV 296	return Err(eyre!("unexpected stream item {:?}", item));	×
297	};
298	assert!(sts.unwrap().success());	1✔
299	assert_closed(&mut stream).await;	1✔
300	Ok(())	2✔
301	}	1✔
302
303	#[tokio::test]
304	async fn test_kill() -> Result<()> {	1✔
305	let mut stream = stream_cmd(&["/bin/sh", "-c", "sleep 60"]).await?;	1✔
306	if let Some(child) = stream.child_mut() {	1✔
307	let _ = child.start_kill();	1✔
308	}	1✔
309	let item = stream_next(&mut stream).await?;	1✔
310	assert_eq!(item, Item::Done(Ok(ExitSts::Signal(9))));	1✔
311	assert_closed(&mut stream).await;	1✔
312	Ok(())	2✔
313	}	1✔
314
315	#[test]
316	fn test_exitsts_display() {	1✔
317	assert_eq!(format!("{}", ExitSts::Success), "success");	1✔
318	assert_eq!(format!("{}", ExitSts::Code(42)), "code 42");	1✔
319	assert_eq!(format!("{}", ExitSts::Signal(1)), "signal 1 (SIGHUP)");	1✔
320	assert_eq!(format!("{}", ExitSts::Signal(12345)), "signal 12345");	1✔
321	}	1✔
322	}

lpenz / ogle / 29192566528

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