tokio_process_tools/process/
builder.rs

1use super::name::{ProcessName, generate_name};
2use super::stream_config::{ProcessStreamBuilder, ProcessStreamConfig};
3use crate::error::SpawnError;
4use crate::output_stream::OutputStream;
5use crate::process_handle::ProcessHandle;
6use std::marker::PhantomData;
7
8#[doc(hidden)]
9pub struct Unnamed;
10
11#[doc(hidden)]
12pub struct Named {
13    name: ProcessName,
14}
15
16#[doc(hidden)]
17pub struct Unset;
18
19/// Typestate builder for configuring and spawning a process.
20///
21/// A process must be named before configuring output streams. This keeps public errors and tracing
22/// fields intentional, while stdout and stderr stream configuration remains explicit at the spawn
23/// call site.
24///
25/// # Examples
26///
27/// ```no_run
28/// use tokio_process_tools::*;
29/// use tokio_process_tools::SpawnError;
30/// use tokio::process::Command;
31///
32/// # tokio_test::block_on(async {
33/// let process = Process::new(Command::new("cargo"))
34///     .name("test-runner")
35///     .stdout_and_stderr(|stream| {
36///         stream
37///             .broadcast()
38///             .best_effort_delivery()
39///             .no_replay()
40///             .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
41///             .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
42///     })
43///     .spawn()?;
44/// # Ok::<_, SpawnError>(())
45/// # });
46/// ```
47pub struct Process<
48    NameState = Unnamed,
49    StdoutConfig = Unset,
50    Stdout = Unset,
51    StderrConfig = Unset,
52    Stderr = Unset,
53> {
54    cmd: tokio::process::Command,
55    name_state: NameState,
56    stdout_config: StdoutConfig,
57    stderr_config: StderrConfig,
58    _streams: PhantomData<fn() -> (Stdout, Stderr)>,
59}
60
61impl Process {
62    /// Creates a new process builder from a tokio command.
63    #[must_use]
64    pub fn new(cmd: tokio::process::Command) -> Self {
65        Self {
66            cmd,
67            name_state: Unnamed,
68            stdout_config: Unset,
69            stderr_config: Unset,
70            _streams: PhantomData,
71        }
72    }
73
74    /// Sets how the process should be named.
75    ///
76    /// You can provide either an explicit name or configure automatic name generation.
77    /// The name is used in public errors and tracing fields. By default, automatic
78    /// naming captures only the program name. Prefer `.name(...)` for stable safe
79    /// labels when command arguments or environment variables may contain secrets.
80    #[must_use]
81    pub fn name(self, name: impl Into<ProcessName>) -> Process<Named> {
82        Process {
83            cmd: self.cmd,
84            name_state: Named { name: name.into() },
85            stdout_config: Unset,
86            stderr_config: Unset,
87            _streams: PhantomData,
88        }
89    }
90}
91
92impl Process<Named> {
93    /// Configures stdout and stderr with the same output stream settings.
94    #[must_use]
95    pub fn stdout_and_stderr<Config, Stream>(
96        self,
97        configure: impl FnOnce(ProcessStreamBuilder) -> Config,
98    ) -> Process<Named, Config, Stream, Config, Stream>
99    where
100        Config: ProcessStreamConfig<Stream> + Copy,
101        Stream: OutputStream,
102    {
103        let config = configure(ProcessStreamBuilder);
104        Process {
105            cmd: self.cmd,
106            name_state: self.name_state,
107            stdout_config: config,
108            stderr_config: config,
109            _streams: PhantomData,
110        }
111    }
112
113    /// Configures stdout before configuring stderr.
114    #[must_use]
115    pub fn stdout<StdoutConfig, Stdout>(
116        self,
117        configure: impl FnOnce(ProcessStreamBuilder) -> StdoutConfig,
118    ) -> Process<Named, StdoutConfig, Stdout>
119    where
120        StdoutConfig: ProcessStreamConfig<Stdout>,
121        Stdout: OutputStream,
122    {
123        Process {
124            cmd: self.cmd,
125            name_state: self.name_state,
126            stdout_config: configure(ProcessStreamBuilder),
127            stderr_config: Unset,
128            _streams: PhantomData,
129        }
130    }
131
132    /// Configures stderr before configuring stdout.
133    #[must_use]
134    pub fn stderr<StderrConfig, Stderr>(
135        self,
136        configure: impl FnOnce(ProcessStreamBuilder) -> StderrConfig,
137    ) -> Process<Named, Unset, Unset, StderrConfig, Stderr>
138    where
139        StderrConfig: ProcessStreamConfig<Stderr>,
140        Stderr: OutputStream,
141    {
142        Process {
143            cmd: self.cmd,
144            name_state: self.name_state,
145            stdout_config: Unset,
146            stderr_config: configure(ProcessStreamBuilder),
147            _streams: PhantomData,
148        }
149    }
150}
151
152impl<StdoutConfig, Stdout> Process<Named, StdoutConfig, Stdout>
153where
154    Stdout: OutputStream,
155{
156    /// Configures stderr and completes the process builder.
157    #[must_use]
158    pub fn stderr<StderrConfig, Stderr>(
159        self,
160        configure: impl FnOnce(ProcessStreamBuilder) -> StderrConfig,
161    ) -> Process<Named, StdoutConfig, Stdout, StderrConfig, Stderr>
162    where
163        StdoutConfig: ProcessStreamConfig<Stdout>,
164        StderrConfig: ProcessStreamConfig<Stderr>,
165        Stderr: OutputStream,
166    {
167        Process {
168            cmd: self.cmd,
169            name_state: self.name_state,
170            stdout_config: self.stdout_config,
171            stderr_config: configure(ProcessStreamBuilder),
172            _streams: PhantomData,
173        }
174    }
175}
176
177impl<StderrConfig, Stderr> Process<Named, Unset, Unset, StderrConfig, Stderr>
178where
179    Stderr: OutputStream,
180{
181    /// Configures stdout and completes the process builder.
182    #[must_use]
183    pub fn stdout<StdoutConfig, Stdout>(
184        self,
185        configure: impl FnOnce(ProcessStreamBuilder) -> StdoutConfig,
186    ) -> Process<Named, StdoutConfig, Stdout, StderrConfig, Stderr>
187    where
188        StderrConfig: ProcessStreamConfig<Stderr>,
189        StdoutConfig: ProcessStreamConfig<Stdout>,
190        Stdout: OutputStream,
191    {
192        Process {
193            cmd: self.cmd,
194            name_state: self.name_state,
195            stdout_config: configure(ProcessStreamBuilder),
196            stderr_config: self.stderr_config,
197            _streams: PhantomData,
198        }
199    }
200}
201
202impl<StdoutConfig, Stdout, StderrConfig, Stderr>
203    Process<Named, StdoutConfig, Stdout, StderrConfig, Stderr>
204where
205    Stdout: OutputStream,
206    Stderr: OutputStream,
207{
208    /// Spawns the process with the configured output streams.
209    ///
210    /// # Errors
211    ///
212    /// Returns [`SpawnError::SpawnFailed`] if the process cannot be spawned.
213    pub fn spawn(self) -> Result<ProcessHandle<Stdout, Stderr>, SpawnError>
214    where
215        StdoutConfig: ProcessStreamConfig<Stdout>,
216        StderrConfig: ProcessStreamConfig<Stderr>,
217    {
218        let name = generate_name(&self.name_state.name, &self.cmd);
219        ProcessHandle::<Stdout, Stderr>::spawn_with_stream_configs(
220            name,
221            self.cmd,
222            self.stdout_config,
223            self.stderr_config,
224        )
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231    use crate::output_stream::TrySubscribable;
232    use crate::test_support::{ScriptedOutput, line_output_options};
233    use crate::{
234        AutoName, AutoNameSettings, BestEffortDelivery, DEFAULT_MAX_BUFFERED_CHUNKS,
235        DEFAULT_READ_CHUNK_SIZE, NoReplay, NumBytesExt, ProcessHandle, ProcessOutput,
236        ReliableDelivery, ReplayEnabled, SingleSubscriberOutputStream,
237    };
238    use assertr::prelude::*;
239    use std::time::Duration;
240    use tokio::process::Command;
241
242    async fn assert_successful_completion<Stdout, Stderr>(
243        mut process: ProcessHandle<Stdout, Stderr>,
244    ) where
245        Stdout: TrySubscribable,
246        Stderr: TrySubscribable,
247    {
248        let ProcessOutput {
249            status,
250            stdout,
251            stderr,
252        } = process
253            .wait_for_completion_with_output(Duration::from_secs(2), line_output_options())
254            .await
255            .unwrap()
256            .expect_completed("process should complete");
257
258        assert_that!(status.success()).is_true();
259        assert_that!(stdout.lines().is_empty()).is_false();
260        assert_that!(stderr.lines().is_empty()).is_true();
261    }
262
263    async fn assert_out_and_err_completion<Stdout, Stderr>(
264        mut process: ProcessHandle<Stdout, Stderr>,
265    ) where
266        Stdout: TrySubscribable,
267        Stderr: TrySubscribable,
268    {
269        let output = process
270            .wait_for_completion_with_output(Duration::from_secs(2), line_output_options())
271            .await
272            .unwrap()
273            .expect_completed("process should complete");
274
275        assert_that!(output.status.success()).is_true();
276        assert_that!(output.stdout.lines().iter().map(String::as_str)).contains_exactly(["out"]);
277        assert_that!(output.stderr.lines().iter().map(String::as_str)).contains_exactly(["err"]);
278    }
279
280    mod shared_config {
281        use super::*;
282
283        #[tokio::test]
284        async fn shared_broadcast_config_applies_to_stdout_and_stderr() {
285            let process = Process::new(ScriptedOutput::builder().stdout("out\n").build())
286                .name(AutoName::program_only())
287                .stdout_and_stderr(|stream| {
288                    stream
289                        .broadcast()
290                        .best_effort_delivery()
291                        .replay_last_bytes(1.megabytes())
292                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
293                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
294                })
295                .spawn()
296                .expect("Failed to spawn");
297
298            assert_that!(process.stdout().read_chunk_size()).is_equal_to(DEFAULT_READ_CHUNK_SIZE);
299            assert_that!(process.stdout().max_buffered_chunks())
300                .is_equal_to(DEFAULT_MAX_BUFFERED_CHUNKS);
301            assert_that!(process.stderr().read_chunk_size()).is_equal_to(DEFAULT_READ_CHUNK_SIZE);
302            assert_that!(process.stderr().max_buffered_chunks())
303                .is_equal_to(DEFAULT_MAX_BUFFERED_CHUNKS);
304            assert_successful_completion(process).await;
305        }
306
307        #[tokio::test]
308        async fn shared_single_subscriber_config_applies_to_stdout_and_stderr() {
309            let process = Process::new(ScriptedOutput::builder().stdout("out\n").build())
310                .name(AutoName::program_only())
311                .stdout_and_stderr(|stream| {
312                    stream
313                        .single_subscriber()
314                        .best_effort_delivery()
315                        .no_replay()
316                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
317                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
318                })
319                .spawn()
320                .expect("Failed to spawn");
321
322            assert_that!(process.stdout().read_chunk_size()).is_equal_to(DEFAULT_READ_CHUNK_SIZE);
323            assert_that!(process.stdout().max_buffered_chunks())
324                .is_equal_to(DEFAULT_MAX_BUFFERED_CHUNKS);
325            assert_that!(process.stdout().replay_enabled()).is_false();
326            assert_that!(process.stderr().read_chunk_size()).is_equal_to(DEFAULT_READ_CHUNK_SIZE);
327            assert_that!(process.stderr().max_buffered_chunks())
328                .is_equal_to(DEFAULT_MAX_BUFFERED_CHUNKS);
329            assert_that!(process.stderr().replay_enabled()).is_false();
330            assert_successful_completion(process).await;
331        }
332    }
333
334    mod split_config {
335        use super::*;
336
337        #[tokio::test]
338        async fn split_broadcast_config_applies_per_stream() {
339            let process = Process::new(ScriptedOutput::builder().stdout("out\n").build())
340                .name(AutoName::program_only())
341                .stdout(|stdout| {
342                    stdout
343                        .broadcast()
344                        .best_effort_delivery()
345                        .replay_last_bytes(1.megabytes())
346                        .read_chunk_size(42.kilobytes())
347                        .max_buffered_chunks(42)
348                })
349                .stderr(|stderr| {
350                    stderr
351                        .broadcast()
352                        .best_effort_delivery()
353                        .replay_last_bytes(1.megabytes())
354                        .read_chunk_size(43.kilobytes())
355                        .max_buffered_chunks(43)
356                })
357                .spawn()
358                .expect("Failed to spawn");
359
360            assert_that!(process.stdout().read_chunk_size()).is_equal_to(42.kilobytes());
361            assert_that!(process.stdout().max_buffered_chunks()).is_equal_to(42);
362            assert_that!(process.stderr().read_chunk_size()).is_equal_to(43.kilobytes());
363            assert_that!(process.stderr().max_buffered_chunks()).is_equal_to(43);
364            assert_successful_completion(process).await;
365        }
366
367        #[tokio::test]
368        async fn split_single_subscriber_config_applies_per_stream() {
369            let process = Process::new(ScriptedOutput::builder().stdout("out\n").build())
370                .name(AutoName::program_only())
371                .stdout(|stdout| {
372                    stdout
373                        .single_subscriber()
374                        .best_effort_delivery()
375                        .replay_last_bytes(1.megabytes())
376                        .read_chunk_size(42.kilobytes())
377                        .max_buffered_chunks(42)
378                })
379                .stderr(|stderr| {
380                    stderr
381                        .single_subscriber()
382                        .best_effort_delivery()
383                        .replay_last_bytes(1.megabytes())
384                        .read_chunk_size(43.kilobytes())
385                        .max_buffered_chunks(43)
386                })
387                .spawn()
388                .expect("Failed to spawn");
389
390            assert_that!(process.stdout().read_chunk_size()).is_equal_to(42.kilobytes());
391            assert_that!(process.stdout().max_buffered_chunks()).is_equal_to(42);
392            assert_that!(process.stderr().read_chunk_size()).is_equal_to(43.kilobytes());
393            assert_that!(process.stderr().max_buffered_chunks()).is_equal_to(43);
394            assert_successful_completion(process).await;
395        }
396
397        #[tokio::test]
398        async fn split_broadcast_config_applies_per_stream_with_dual_outputs() {
399            let process = Process::new(
400                ScriptedOutput::builder()
401                    .stdout("out\n")
402                    .stderr("err\n")
403                    .build(),
404            )
405            .name(AutoName::program_only())
406            .stdout(|stdout| {
407                stdout
408                    .broadcast()
409                    .reliable_for_active_subscribers()
410                    .replay_last_bytes(1.megabytes())
411                    .read_chunk_size(21.bytes())
412                    .max_buffered_chunks(22)
413            })
414            .stderr(|stderr| {
415                stderr
416                    .broadcast()
417                    .reliable_for_active_subscribers()
418                    .replay_last_bytes(1.megabytes())
419                    .read_chunk_size(23.bytes())
420                    .max_buffered_chunks(24)
421            })
422            .spawn()
423            .expect("Failed to spawn");
424
425            assert_that!(process.stdout().read_chunk_size()).is_equal_to(21.bytes());
426            assert_that!(process.stdout().max_buffered_chunks()).is_equal_to(22);
427            assert_that!(process.stderr().read_chunk_size()).is_equal_to(23.bytes());
428            assert_that!(process.stderr().max_buffered_chunks()).is_equal_to(24);
429            assert_out_and_err_completion(process).await;
430        }
431
432        #[tokio::test]
433        async fn split_broadcast_replay_can_be_sealed() {
434            let process = Process::new(
435                ScriptedOutput::builder()
436                    .stdout("out\n")
437                    .stderr("err\n")
438                    .build(),
439            )
440            .name(AutoName::program_only())
441            .stdout(|stdout| {
442                stdout
443                    .broadcast()
444                    .reliable_for_active_subscribers()
445                    .replay_last_bytes(1.megabytes())
446                    .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
447                    .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
448            })
449            .stderr(|stderr| {
450                stderr
451                    .broadcast()
452                    .best_effort_delivery()
453                    .replay_last_bytes(1.megabytes())
454                    .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
455                    .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
456            })
457            .spawn()
458            .expect("Failed to spawn");
459
460            assert_that!(process.stdout().is_replay_sealed()).is_false();
461            process.seal_stdout_replay();
462            assert_that!(process.stdout().is_replay_sealed()).is_true();
463            assert_out_and_err_completion(process).await;
464        }
465
466        #[tokio::test]
467        async fn split_with_broadcast_stdout_and_single_subscriber_stderr_completes() {
468            let process = Process::new(
469                ScriptedOutput::builder()
470                    .stdout("out\n")
471                    .stderr("err\n")
472                    .build(),
473            )
474            .name(AutoName::program_only())
475            .stdout(|stdout| {
476                stdout
477                    .broadcast()
478                    .reliable_for_active_subscribers()
479                    .replay_last_bytes(1.megabytes())
480                    .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
481                    .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
482            })
483            .stderr(|stderr| {
484                stderr
485                    .single_subscriber()
486                    .best_effort_delivery()
487                    .replay_last_bytes(1.megabytes())
488                    .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
489                    .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
490            })
491            .spawn()
492            .expect("Failed to spawn");
493
494            process.seal_stdout_replay();
495            assert_that!(process.stdout().is_replay_sealed()).is_true();
496            assert_out_and_err_completion(process).await;
497        }
498    }
499
500    mod single_subscriber_delivery_and_replay {
501        use super::*;
502
503        fn assert_single_subscriber_stream_types<StdoutD, StdoutR, StderrD, StderrR>(
504            _process: &ProcessHandle<
505                SingleSubscriberOutputStream<StdoutD, StdoutR>,
506                SingleSubscriberOutputStream<StderrD, StderrR>,
507            >,
508        ) where
509            StdoutD: crate::Delivery,
510            StdoutR: crate::Replay,
511            StderrD: crate::Delivery,
512            StderrR: crate::Replay,
513        {
514        }
515
516        #[tokio::test]
517        async fn split_delivery_modes_can_wait_for_completion() {
518            let mut process = Process::new(Command::new("ls"))
519                .name(AutoName::program_only())
520                .stdout(|stdout| {
521                    stdout
522                        .single_subscriber()
523                        .reliable_for_active_subscribers()
524                        .no_replay()
525                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
526                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
527                })
528                .stderr(|stderr| {
529                    stderr
530                        .single_subscriber()
531                        .best_effort_delivery()
532                        .no_replay()
533                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
534                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
535                })
536                .spawn()
537                .expect("Failed to spawn");
538
539            assert_single_subscriber_stream_types::<
540                ReliableDelivery,
541                NoReplay,
542                BestEffortDelivery,
543                NoReplay,
544            >(&process);
545
546            let _ = process
547                .wait_for_completion(Duration::from_secs(2))
548                .await
549                .unwrap();
550        }
551
552        #[tokio::test]
553        async fn split_replay_modes_preserve_stream_types() {
554            let mut process = Process::new(Command::new("ls"))
555                .name(AutoName::program_only())
556                .stdout(|stdout| {
557                    stdout
558                        .single_subscriber()
559                        .best_effort_delivery()
560                        .no_replay()
561                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
562                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
563                })
564                .stderr(|stderr| {
565                    stderr
566                        .single_subscriber()
567                        .reliable_for_active_subscribers()
568                        .replay_last_chunks(1)
569                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
570                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
571                })
572                .spawn()
573                .expect("Failed to spawn");
574
575            assert_single_subscriber_stream_types::<
576                BestEffortDelivery,
577                NoReplay,
578                ReliableDelivery,
579                ReplayEnabled,
580            >(&process);
581
582            process.seal_stderr_replay();
583            let _ = process
584                .wait_for_completion(Duration::from_secs(2))
585                .await
586                .unwrap();
587        }
588
589        #[tokio::test]
590        async fn split_replay_enabled_streams_can_be_sealed() {
591            let mut process = Process::new(Command::new("ls"))
592                .name(AutoName::program_only())
593                .stdout(|stdout| {
594                    stdout
595                        .single_subscriber()
596                        .best_effort_delivery()
597                        .replay_last_chunks(1)
598                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
599                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
600                })
601                .stderr(|stderr| {
602                    stderr
603                        .single_subscriber()
604                        .reliable_for_active_subscribers()
605                        .replay_last_chunks(1)
606                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
607                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
608                })
609                .spawn()
610                .expect("Failed to spawn");
611
612            assert_that!(process.stdout().replay_enabled()).is_true();
613            assert_that!(process.stderr().replay_enabled()).is_true();
614
615            process.seal_output_replay();
616            assert_that!(process.stdout().is_replay_sealed()).is_true();
617            assert_that!(process.stderr().is_replay_sealed()).is_true();
618
619            let _ = process
620                .wait_for_completion(Duration::from_secs(2))
621                .await
622                .unwrap();
623        }
624    }
625
626    mod discard {
627        use super::*;
628        use crate::OutputStream;
629
630        #[tokio::test]
631        async fn stdout_and_stderr_complete_with_wait_for_completion() {
632            let mut process = Process::new(
633                ScriptedOutput::builder()
634                    .stdout("out\n")
635                    .stderr("err\n")
636                    .build(),
637            )
638            .name(AutoName::program_only())
639            .stdout_and_stderr(ProcessStreamBuilder::discard)
640            .spawn()
641            .expect("Failed to spawn");
642
643            assert_that!(process.stdout().name()).is_equal_to("stdout");
644            assert_that!(process.stderr().name()).is_equal_to("stderr");
645
646            process
647                .wait_for_completion(Duration::from_secs(2))
648                .await
649                .unwrap()
650                .expect_completed("process should complete");
651        }
652
653        #[tokio::test]
654        async fn can_discard_stdout_and_broadcast_stderr() {
655            let mut process = Process::new(
656                ScriptedOutput::builder()
657                    .stdout("out\n")
658                    .stderr("err\n")
659                    .build(),
660            )
661            .name(AutoName::program_only())
662            .stdout(ProcessStreamBuilder::discard)
663            .stderr(|stream| {
664                stream
665                    .broadcast()
666                    .best_effort_delivery()
667                    .replay_last_bytes(1.megabytes())
668                    .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
669                    .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
670            })
671            .spawn()
672            .expect("Failed to spawn");
673
674            let collector = process.stderr().collect_lines_into_vec(
675                crate::LineParsingOptions::default(),
676                crate::test_support::line_collection_options(),
677            );
678
679            process
680                .wait_for_completion(Duration::from_secs(2))
681                .await
682                .unwrap()
683                .expect_completed("process should complete");
684
685            let collected = collector.wait().await.unwrap();
686            assert_that!(collected.lines()).contains_exactly(["err"]);
687        }
688
689        #[tokio::test]
690        async fn can_broadcast_stdout_and_discard_stderr() {
691            let mut process = Process::new(
692                ScriptedOutput::builder()
693                    .stdout("out\n")
694                    .stderr("err\n")
695                    .build(),
696            )
697            .name(AutoName::program_only())
698            .stdout(|stream| {
699                stream
700                    .broadcast()
701                    .best_effort_delivery()
702                    .replay_last_bytes(1.megabytes())
703                    .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
704                    .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
705            })
706            .stderr(ProcessStreamBuilder::discard)
707            .spawn()
708            .expect("Failed to spawn");
709
710            let collector = process.stdout().collect_lines_into_vec(
711                crate::LineParsingOptions::default(),
712                crate::test_support::line_collection_options(),
713            );
714
715            process
716                .wait_for_completion(Duration::from_secs(2))
717                .await
718                .unwrap()
719                .expect_completed("process should complete");
720
721            let collected = collector.wait().await.unwrap();
722            assert_that!(collected.lines()).contains_exactly(["out"]);
723        }
724    }
725
726    mod spawn_errors {
727        use super::*;
728
729        #[tokio::test]
730        async fn default_auto_name_does_not_capture_sensitive_args_in_spawn_error() {
731            let sensitive_arg = "--token=secret-token-should-not-be-logged";
732            let mut cmd = Command::new("tokio-process-tools-definitely-missing-command");
733            cmd.arg(sensitive_arg);
734
735            let error = match Process::new(cmd)
736                .name(AutoName::program_only())
737                .stdout_and_stderr(|stream| {
738                    stream
739                        .broadcast()
740                        .best_effort_delivery()
741                        .no_replay()
742                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
743                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
744                })
745                .spawn()
746            {
747                Ok(mut process) => {
748                    let _ = process.wait_for_completion(Duration::from_secs(2)).await;
749                    assert_that!(()).fail("command should fail to spawn");
750                    return;
751                }
752                Err(error) => error,
753            };
754            let error = error.to_string();
755
756            assert_that!(error.as_str()).contains("tokio-process-tools-definitely-missing-command");
757            assert_that!(error.as_str()).does_not_contain(sensitive_arg);
758        }
759    }
760
761    mod names {
762        use super::*;
763
764        #[tokio::test]
765        async fn auto_name_settings_include_current_dir_and_args() {
766            let mut cmd = Command::new("ls");
767            cmd.arg("-la");
768            cmd.env("IGNORED_ENV", "secret");
769            cmd.current_dir("./");
770
771            let mut process = Process::new(cmd)
772                .name(
773                    AutoNameSettings::builder()
774                        .include_current_dir(true)
775                        .include_args(true)
776                        .build(),
777                )
778                .stdout_and_stderr(|stream| {
779                    stream
780                        .broadcast()
781                        .best_effort_delivery()
782                        .no_replay()
783                        .read_chunk_size(DEFAULT_READ_CHUNK_SIZE)
784                        .max_buffered_chunks(DEFAULT_MAX_BUFFERED_CHUNKS)
785                })
786                .spawn()
787                .expect("Failed to spawn");
788
789            assert_that!(&process.name).is_equal_to("./ % ls \"-la\"");
790
791            let _ = process.wait_for_completion(Duration::from_secs(2)).await;
792        }
793    }
794}
tokio_process_tools/process/builder.rs

tokio_process_tools/process/
builder.rs
