xianyun_executor/exec/
mod.rs

1//! Subprocess execution with logging.
2//!
3//! Ported from `internal/executor/exec.go`, `internal/executor/client.go`,
4//! `internal/executor/interface.go`, and `internal/executor/context.go`.
5//!
6//! Real subprocess execution uses `tokio::process::Command` internally,
7//! bridged from the synchronous [`CommandRunner`] trait via a lazily
8//! started background `tokio::runtime::Runtime` (see `runtime`).
9//! Cancellation/timeouts race the child's `wait()` against
10//! `tokio::time::sleep` with `tokio::select!`, killing the child with
11//! `Child::kill()` on timeout — see `PORT_NOTES.md` for the full
12//! architecture write-up and its known limitations.
13
14pub mod context;
15pub mod mise_decorator;
16pub mod noop;
17pub mod repository;
18pub mod streamed;
19
20#[cfg(test)]
21pub(crate) mod mock;
22
23use std::io::Read;
24use std::path::PathBuf;
25use std::process::Stdio;
26use std::sync::{Arc, Mutex};
27use std::time::{Duration, Instant};
28
29use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
30use tokio::process::Command as TokioCommand;
31
32pub use context::{ErrRunnerMissing, from_context, with_context};
33pub use noop::NoopClient;
34pub use repository::{RepositoryExecutorDecorator, get_repository_decorators};
35pub use streamed::ExecStreamedOpts;
36
37/// Errors ported from the sentinel `errors.New`/`fmt.Errorf` values in
38/// `exec.go`.
39///
40/// Named `ExecError` (rather than `Error`, which Go's `error` interface
41/// doesn't require naming) to leave `std::result::Result`'s `Err` type
42/// unambiguous when this crate's own `ExecResult` struct is also in scope.
43#[derive(Debug, thiserror::Error)]
44pub enum ExecError {
45    /// Ported from `ErrCommandFailed`.
46    #[error("command exited with non-zero status: {0}")]
47    CommandFailed(i32),
48    /// Ported from the `fmt.Errorf("running command: %w", ...)` wrap around
49    /// a `go-cmd`/`os/exec` start failure (binary missing, permission
50    /// denied, etc).
51    #[error("running command: {0}")]
52    RunFailed(String),
53    /// Ported from context cancellation overriding the result (`ctx.Err()`
54    /// checked after the command finishes).
55    #[error("command cancelled")]
56    Cancelled,
57    /// Ported from `LookPath`'s wrapped `exec.LookPath` error.
58    #[error("looking up executable: {0}")]
59    LookPathFailed(String),
60}
61
62/// Ported from `Result` in `exec.go`. Renamed to avoid shadowing
63/// `std::result::Result`.
64#[derive(Debug, Default)]
65pub struct ExecResult {
66    pub stdout: Vec<String>,
67    pub stderr: Vec<String>,
68    pub exit_code: i32,
69    pub error: Option<ExecError>,
70    pub duration: Duration,
71}
72
73impl ExecResult {
74    /// Ported from `Result.CombinedOutput`.
75    pub fn combined_output(&self) -> String {
76        let mut output = String::new();
77        for line in &self.stdout {
78            output.push_str(line);
79            output.push('\n');
80        }
81        for line in &self.stderr {
82            output.push_str(line);
83            output.push('\n');
84        }
85        output
86    }
87}
88
89/// Log level for streamed/buffered output. Ported from `slog.Level` usage
90/// in `Options`; this crate uses `tracing::Level` as the idiomatic Rust
91/// equivalent of Go's `log/slog` (see `PORT_NOTES.md`).
92pub type LogLevel = tracing::Level;
93
94/// Ported from `Options` in `exec.go`.
95///
96/// `timeout` has no Go counterpart on `Options` itself — Go threads a
97/// deadline through the ambient `context.Context` argument every function
98/// in this package takes. This port has no ambient context (see the
99/// "Design notes" section of `PORT_NOTES.md`), so a deadline is passed
100/// explicitly here instead. This is a disclosed deviation, not a silent
101/// behavioral gap.
102///
103/// `log_writer` corresponds to Go's `Logger *slog.Logger` field. Go's
104/// tests inject a `*slog.Logger` writing to an in-memory buffer
105/// (`captureLogger`) to assert on the single "command error"/"command
106/// completed" completion record `logRunCompletion` emits after a run
107/// finishes. This port doesn't thread a full `tracing::Dispatch` through
108/// here (that would need `tracing::dispatcher::with_default`'s
109/// thread-local guard held across the `.await` points in
110/// `run_async`, which doesn't survive a task being moved between
111/// worker threads) -- instead `log_writer`, when set, receives that same
112/// one completion line, formatted directly by `log_completion`. When
113/// `None` (the default), no completion line is emitted at all, which is
114/// a disclosed narrowing from Go's "falls back to `slog.Default()`"
115/// behavior: this port has no equivalent of a mutable, ambient global
116/// default logger to fall back to (see this crate's other "no
117/// reassignable package-level state" deviations), and no caller outside
118/// the ported tests currently needs one.
119#[derive(Default)]
120pub struct Options {
121    pub log_level: Option<LogLevel>,
122    pub stderr_log_level: Option<LogLevel>,
123    pub failure_log_level: Option<LogLevel>,
124    pub disable_streaming: bool,
125    pub stdin: Option<Box<dyn Read + Send>>,
126    pub env: Option<Vec<String>>,
127    pub timeout: Option<Duration>,
128    pub log_writer: Option<Arc<Mutex<Vec<u8>>>>,
129}
130
131/// Ported from `logRunCompletion`. Writes a single completion line to
132/// `opts.log_writer` (if set) describing whether `name`/`args` finished
133/// with `result.error` or completed cleanly, at `failure_log_level`
134/// (default `ERROR`) or `log_level` (default `DEBUG`) respectively.
135/// A no-op when `opts.log_writer` is `None`.
136fn log_completion(opts: &Options, name: &str, args: &[String], result: &ExecResult) {
137    let Some(writer) = &opts.log_writer else {
138        return;
139    };
140
141    let (level, message) = match &result.error {
142        Some(error) => (
143            opts.failure_log_level.unwrap_or(LogLevel::ERROR),
144            format!("command error: {error}"),
145        ),
146        None => (
147            opts.log_level.unwrap_or(LogLevel::DEBUG),
148            "command completed".to_string(),
149        ),
150    };
151
152    let line = format!(
153        "level={level} msg=\"{message}\" command={name} args={args:?} exit_code={exit_code}\n",
154        exit_code = result.exit_code,
155    );
156
157    if let Ok(mut buf) = writer.lock() {
158        let _ = std::io::Write::write_all(&mut *buf, line.as_bytes());
159    }
160}
161
162/// Ported from the `CommandRunner` interface in `interface.go`.
163///
164/// Deliberately kept fully synchronous (see `PORT_NOTES.md`'s Phase 4
165/// architecture note): every implementation below bridges to
166/// `tokio::process::Command` internally via `runtime` +
167/// `Runtime::block_on`, so callers (and the cross-crate consumers in
168/// `xianyun-testutil`/`xianyun-backends`/`xianyun-cluster`/
169/// `xianyun-deploy`) see no trait signature change.
170pub trait CommandRunner {
171    fn run(&self, name: &str, args: &[String], opts: Option<Options>) -> ExecResult;
172
173    fn read_stdout(&self, name: &str, args: &[String]) -> Result<String, ExecError>;
174
175    fn run_sudo(&self, name: &str, args: &[String]) -> Result<(), ExecError>;
176
177    fn check(&self, name: &str, args: &[String]) -> Result<bool, ExecError>;
178
179    fn exec_streamed_command(
180        &self,
181        opts: Option<ExecStreamedOpts>,
182        name: &str,
183        args: &[String],
184    ) -> Result<(), ExecError>;
185
186    fn look_path(&self, file: &str) -> Result<PathBuf, ExecError>;
187
188    fn run_in_repository(
189        &self,
190        repo_dir: &str,
191        name: &str,
192        args: &[String],
193        opts: Option<ExecStreamedOpts>,
194    ) -> Result<(), ExecError>;
195}
196
197/// Returns a lazily started, process-wide background Tokio runtime used to
198/// bridge the synchronous [`CommandRunner`] trait to `tokio::process`.
199///
200/// # Known limitation
201///
202/// Calling any function in this module from code that is *already*
203/// running inside a Tokio runtime (e.g. an `async fn` body) will panic
204/// with "Cannot start a runtime from within a runtime", because
205/// `Runtime::block_on` refuses to nest regardless of which `Runtime`
206/// instance is used. No caller in this workspace does that today; if one
207/// needs to in the future, wrap the call in `tokio::task::spawn_blocking`
208/// (or migrate `CommandRunner` to `async fn`, which is the cross-crate
209/// breaking change this task's `PORT_NOTES.md` explicitly deferred).
210pub(crate) fn runtime() -> &'static tokio::runtime::Runtime {
211    static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
212    RUNTIME.get_or_init(|| {
213        tokio::runtime::Runtime::new().expect("failed to start xianyun-executor background runtime")
214    })
215}
216
217/// Ported from `Client` in `client.go`.
218pub struct Client;
219
220impl Client {
221    pub fn new() -> Self {
222        Client
223    }
224}
225
226impl Default for Client {
227    fn default() -> Self {
228        Self::new()
229    }
230}
231
232impl CommandRunner for Client {
233    fn run(&self, name: &str, args: &[String], opts: Option<Options>) -> ExecResult {
234        run(name, args, opts)
235    }
236
237    fn read_stdout(&self, name: &str, args: &[String]) -> Result<String, ExecError> {
238        read_stdout(name, args)
239    }
240
241    fn run_sudo(&self, name: &str, args: &[String]) -> Result<(), ExecError> {
242        run_sudo(name, args)
243    }
244
245    fn check(&self, name: &str, args: &[String]) -> Result<bool, ExecError> {
246        check(name, args)
247    }
248
249    fn exec_streamed_command(
250        &self,
251        opts: Option<ExecStreamedOpts>,
252        name: &str,
253        args: &[String],
254    ) -> Result<(), ExecError> {
255        streamed::exec_streamed_command(opts, name, args)
256    }
257
258    fn look_path(&self, file: &str) -> Result<PathBuf, ExecError> {
259        look_path(file)
260    }
261
262    fn run_in_repository(
263        &self,
264        repo_dir: &str,
265        name: &str,
266        args: &[String],
267        opts: Option<ExecStreamedOpts>,
268    ) -> Result<(), ExecError> {
269        let mut opts = opts.unwrap_or_default();
270        if opts.working_dir.as_deref().unwrap_or_default().is_empty() {
271            opts.working_dir = Some(repo_dir.to_string());
272        }
273
274        let mut command = repository::Command {
275            name: name.to_string(),
276            args: args.to_vec(),
277            env: opts.env.clone(),
278            dir: opts.working_dir.clone(),
279        };
280
281        for decorator in get_repository_decorators() {
282            command = decorator.wrap(command)?;
283        }
284
285        let mut decorated_opts = opts;
286        decorated_opts.env = command.env;
287        decorated_opts.working_dir = command.dir;
288
289        streamed::exec_streamed_command(Some(decorated_opts), &command.name, &command.args)
290    }
291}
292
293/// Reads bytes from a piped child stdout/stderr handle line-by-line until
294/// EOF, collecting each line. Spawned as a background task so the pipe is
295/// drained continuously while the child runs — see `PORT_NOTES.md` point 2
296/// ("go-cmd's buffered/streaming split") for why this matters: an
297/// undrained pipe fills the OS buffer (typically 64KB) and blocks the
298/// child's next write forever.
299pub(crate) async fn drain_lines<R>(reader: R) -> Vec<String>
300where
301    R: tokio::io::AsyncRead + Unpin,
302{
303    let mut lines = Vec::new();
304    let mut reader = BufReader::new(reader).lines();
305    while let Ok(Some(line)) = reader.next_line().await {
306        lines.push(line);
307    }
308    lines
309}
310
311/// Applies `opts.env` (replacing the child's entire environment, matching
312/// Go's `command.Env = opts.Env` semantics) to a `tokio::process::Command`.
313pub(crate) fn apply_env(cmd: &mut TokioCommand, env: &Option<Vec<String>>) {
314    if let Some(env) = env {
315        cmd.env_clear();
316        for kv in env {
317            if let Some((key, value)) = kv.split_once('=') {
318                cmd.env(key, value);
319            }
320        }
321    }
322}
323
324/// Core async implementation behind [`run`]. Spawns the child with piped
325/// stdout/stderr, drains both concurrently via [`drain_lines`], and races
326/// `Child::wait()` against `opts.timeout` (if set) with `tokio::select!`,
327/// killing the child on timeout. See `PORT_NOTES.md` for why this
328/// architecture (rather than Go's `context.CommandContext`, which has no
329/// direct Rust equivalent) was chosen.
330async fn run_async(name: &str, args: &[String], mut opts: Options) -> ExecResult {
331    let start = Instant::now();
332
333    let mut cmd = TokioCommand::new(name);
334    cmd.args(args);
335    apply_env(&mut cmd, &opts.env);
336    cmd.stdout(Stdio::piped());
337    cmd.stderr(Stdio::piped());
338    cmd.stdin(if opts.stdin.is_some() {
339        Stdio::piped()
340    } else {
341        Stdio::null()
342    });
343
344    let mut child = match cmd.spawn() {
345        Ok(child) => child,
346        Err(error) => {
347            let result = ExecResult {
348                stdout: Vec::new(),
349                stderr: Vec::new(),
350                exit_code: -1,
351                error: Some(ExecError::RunFailed(error.to_string())),
352                duration: start.elapsed(),
353            };
354            log_completion(&opts, name, args, &result);
355            return result;
356        }
357    };
358
359    if let Some(mut stdin_reader) = opts.stdin.take()
360        && let Some(mut child_stdin) = child.stdin.take()
361    {
362        let mut buf = Vec::new();
363        let _ = stdin_reader.read_to_end(&mut buf);
364        let _ = child_stdin.write_all(&buf).await;
365        drop(child_stdin);
366    }
367
368    let stdout = child.stdout.take().expect("stdout was piped");
369    let stderr = child.stderr.take().expect("stderr was piped");
370    let stdout_task = tokio::spawn(drain_lines(stdout));
371    let stderr_task = tokio::spawn(drain_lines(stderr));
372
373    let wait_result = match opts.timeout {
374        Some(timeout) => {
375            tokio::select! {
376                res = child.wait() => Some(res),
377                () = tokio::time::sleep(timeout) => None,
378            }
379        }
380        None => Some(child.wait().await),
381    };
382
383    let cancelled = wait_result.is_none();
384    let final_status = match wait_result {
385        Some(res) => res,
386        None => {
387            let _ = child.kill().await;
388            child.wait().await
389        }
390    };
391
392    let stdout_lines = stdout_task.await.unwrap_or_default();
393    let stderr_lines = stderr_task.await.unwrap_or_default();
394    let duration = start.elapsed();
395
396    let mut exit_code = -1;
397    let mut error = None;
398
399    match final_status {
400        Ok(status) => {
401            exit_code = status.code().unwrap_or(-1);
402            if !status.success() {
403                error = Some(ExecError::CommandFailed(exit_code));
404            }
405        }
406        Err(run_error) => {
407            error = Some(ExecError::RunFailed(run_error.to_string()));
408        }
409    }
410
411    if cancelled {
412        error = Some(ExecError::Cancelled);
413    }
414
415    let result = ExecResult {
416        stdout: stdout_lines,
417        stderr: stderr_lines,
418        exit_code,
419        error,
420        duration,
421    };
422    log_completion(&opts, name, args, &result);
423    result
424}
425
426/// Ported from the package-level `run` function.
427pub fn run(name: &str, args: &[String], opts: Option<Options>) -> ExecResult {
428    runtime().block_on(run_async(name, args, opts.unwrap_or_default()))
429}
430
431/// Ported from the package-level `readStdout` function.
432pub fn read_stdout(name: &str, args: &[String]) -> Result<String, ExecError> {
433    let result = run(
434        name,
435        args,
436        Some(Options {
437            log_level: Some(LogLevel::DEBUG),
438            stderr_log_level: Some(LogLevel::DEBUG),
439            failure_log_level: Some(LogLevel::DEBUG),
440            disable_streaming: true,
441            ..Default::default()
442        }),
443    );
444
445    if let Some(error) = result.error {
446        return Err(error);
447    }
448
449    Ok(result.stdout.join("\n"))
450}
451
452/// Ported from the package-level `check` function, with no deadline. See
453/// [`check_with_timeout`] for the deviation that threads one through
454/// explicitly (Go's version takes a deadline via the ambient
455/// `context.Context` argument every caller already has).
456pub fn check(name: &str, args: &[String]) -> Result<bool, ExecError> {
457    check_with_timeout(name, args, None)
458}
459
460/// Ported from the package-level `check` function. See `PORT_NOTES.md`:
461/// this crate has no ambient context to carry Go's caller-supplied
462/// deadline, so `timeout` is threaded through explicitly as a disclosed
463/// deviation from a literal 1:1 signature match.
464pub fn check_with_timeout(
465    name: &str,
466    args: &[String],
467    timeout: Option<Duration>,
468) -> Result<bool, ExecError> {
469    let result = run(
470        name,
471        args,
472        Some(Options {
473            log_level: Some(LogLevel::DEBUG),
474            stderr_log_level: Some(LogLevel::DEBUG),
475            failure_log_level: Some(LogLevel::DEBUG),
476            disable_streaming: true,
477            timeout,
478            ..Default::default()
479        }),
480    );
481
482    match result.error {
483        None => Ok(true),
484        Some(ExecError::CommandFailed(_)) => Ok(false),
485        Some(other) => Err(other),
486    }
487}
488
489/// Ported from `runSudo`, wired to the real environment via [`apply_sudo`].
490pub fn run_sudo(name: &str, args: &[String]) -> Result<(), ExecError> {
491    run_sudo_with(name, args, real_geteuid, look_path)
492}
493
494/// Ported from `runSudo`. See [`apply_sudo_with`]'s doc comment: this port
495/// takes the euid getter and PATH lookup as explicit parameters (rather
496/// than Go's reassignable package-level `var`s) so tests can stub them
497/// without touching real root privileges or the real `sudo` binary. Runs
498/// synchronously with inherited stdio (`std::process::Command`), matching
499/// Go's `runSudo`, which deliberately does not capture output so `sudo`
500/// can prompt on the controlling terminal.
501pub fn run_sudo_with(
502    name: &str,
503    args: &[String],
504    geteuid: impl Fn() -> u32,
505    look_path: impl Fn(&str) -> Result<PathBuf, ExecError>,
506) -> Result<(), ExecError> {
507    let (name, args) = apply_sudo_with(name, args, geteuid, look_path);
508
509    let status = std::process::Command::new(&name)
510        .args(&args)
511        .stdin(Stdio::inherit())
512        .stdout(Stdio::inherit())
513        .stderr(Stdio::inherit())
514        .status();
515
516    match status {
517        Ok(status) if status.success() => Ok(()),
518        Ok(status) => match status.code() {
519            Some(code) if code > 0 => Err(ExecError::CommandFailed(code)),
520            _ => Err(ExecError::RunFailed(format!(
521                "process terminated: {status}"
522            ))),
523        },
524        Err(error) => Err(ExecError::RunFailed(error.to_string())),
525    }
526}
527
528/// Ported from `ApplySudo`. Go stubs `geteuid`/`lookPath` as reassignable
529/// package-level vars for testing; Rust has no equivalent to mutable
530/// top-level function pointers without `unsafe` global state, so this port
531/// takes the euid getter and PATH lookup as explicit parameters instead.
532/// The zero-argument `apply_sudo` below is the production entry point,
533/// wiring in a real `geteuid(2)` call and [`look_path`]; ported tests call
534/// `apply_sudo_with` directly. See `PORT_NOTES.md`.
535pub fn apply_sudo_with(
536    name: &str,
537    args: &[String],
538    geteuid: impl Fn() -> u32,
539    look_path: impl Fn(&str) -> Result<PathBuf, ExecError>,
540) -> (String, Vec<String>) {
541    if geteuid() == 0 {
542        return (name.to_string(), args.to_vec());
543    }
544
545    match look_path("sudo") {
546        Ok(sudo_path) => {
547            let mut new_args = vec![name.to_string()];
548            new_args.extend(args.iter().cloned());
549            (sudo_path.to_string_lossy().into_owned(), new_args)
550        }
551        Err(_) => {
552            tracing::warn!(
553                command = name,
554                "sudo requested but not found on PATH; running command directly"
555            );
556            (name.to_string(), args.to_vec())
557        }
558    }
559}
560
561/// Ported from `ApplySudo`, wired to the real environment.
562pub fn apply_sudo(name: &str, args: &[String]) -> (String, Vec<String>) {
563    apply_sudo_with(name, args, real_geteuid, look_path)
564}
565
566/// Real `geteuid(2)` call, avoiding a dependency on the `nix`/`libc` crates
567/// for this single narrow use (the C symbol is already linked into every
568/// Unix Rust binary via the platform's libc).
569#[cfg(unix)]
570fn real_geteuid() -> u32 {
571    unsafe extern "C" {
572        fn geteuid() -> u32;
573    }
574    // SAFETY: `geteuid(2)` takes no arguments, has no preconditions, and
575    // cannot fail -- it's a pure syscall wrapper reading the calling
576    // process's own effective UID.
577    unsafe { geteuid() }
578}
579
580#[cfg(not(unix))]
581fn real_geteuid() -> u32 {
582    // Non-Unix platforms have no euid concept; treat as "never root" so
583    // `apply_sudo` always attempts to elevate via `sudo` (which will
584    // simply not be found on PATH there).
585    1
586}
587
588/// Ported from `lookPath` (aliased to `exec.LookPath` in Go). Minimal
589/// manual `$PATH` search, matching `mise_decorator`'s existing approach of
590/// avoiding a dependency on an external `which`-style crate for this
591/// narrow use.
592pub fn look_path(file: &str) -> Result<PathBuf, ExecError> {
593    if file.contains(std::path::MAIN_SEPARATOR) {
594        let candidate = PathBuf::from(file);
595        return if is_executable_file(&candidate) {
596            Ok(candidate)
597        } else {
598            Err(ExecError::LookPathFailed(file.to_string()))
599        };
600    }
601
602    let path_var =
603        std::env::var_os("PATH").ok_or_else(|| ExecError::LookPathFailed(file.to_string()))?;
604
605    for dir in std::env::split_paths(&path_var) {
606        let candidate = dir.join(file);
607        if is_executable_file(&candidate) {
608            return Ok(candidate);
609        }
610    }
611
612    Err(ExecError::LookPathFailed(file.to_string()))
613}
614
615#[cfg(unix)]
616fn is_executable_file(path: &std::path::Path) -> bool {
617    use std::os::unix::fs::PermissionsExt;
618    std::fs::metadata(path)
619        .is_ok_and(|meta| meta.is_file() && meta.permissions().mode() & 0o111 != 0)
620}
621
622#[cfg(not(unix))]
623fn is_executable_file(path: &std::path::Path) -> bool {
624    path.is_file()
625}
626
627#[cfg(test)]
628mod tests {
629    use super::*;
630    use std::sync::{Arc, Mutex};
631    use std::time::Duration as StdDuration;
632
633    // --- Ported from exec_test.go ---
634
635    #[test]
636    fn run_success() {
637        let result = run("echo", &["hello".into(), "world".into()], None);
638
639        assert!(result.error.is_none(), "command should succeed");
640        assert_eq!(result.exit_code, 0, "exit code should be 0");
641        assert!(!result.stdout.is_empty(), "stdout should contain output");
642        assert!(
643            result.duration > StdDuration::ZERO,
644            "duration should be recorded"
645        );
646    }
647
648    #[test]
649    fn run_command_not_found() {
650        let result = run("nonexistent-command-12345", &[], None);
651
652        assert!(result.error.is_some(), "command should fail");
653        assert_ne!(result.exit_code, 0, "exit code should be non-zero");
654        assert!(
655            result.duration > StdDuration::ZERO,
656            "duration should be recorded"
657        );
658    }
659
660    #[test]
661    fn run_non_zero_exit() {
662        let result = run("sh", &["-c".into(), "exit 42".into()], None);
663
664        assert!(result.error.is_some(), "command should return error");
665        assert_eq!(result.exit_code, 42, "exit code should be 42");
666        assert!(matches!(result.error, Some(ExecError::CommandFailed(42))));
667    }
668
669    #[test]
670    fn run_context_cancellation() {
671        // Go: context.WithTimeout(100ms) around `sleep 10`.
672        //
673        // NOTE: Go's stdlib `context.Context` cancellation is threaded
674        // through `os/exec.CommandContext`, which kills the child process
675        // on cancellation/timeout. This port has no ambient context (see
676        // PORT_NOTES.md), so the 100ms deadline is passed explicitly via
677        // `Options.timeout` instead -- a disclosed deviation from the
678        // literal Go call shape, not a silently dropped test. The
679        // implementation races `tokio::process::Child::wait()` against
680        // `tokio::time::sleep` with `tokio::select!` and calls
681        // `Child::kill()` on timeout.
682        let result = run(
683            "sleep",
684            &["10".into()],
685            Some(Options {
686                timeout: Some(StdDuration::from_millis(100)),
687                ..Default::default()
688            }),
689        );
690
691        assert!(result.error.is_some());
692        assert!(matches!(result.error, Some(ExecError::Cancelled)));
693    }
694
695    #[test]
696    fn run_with_options_custom_logger() {
697        let result = run(
698            "echo",
699            &["test".into()],
700            Some(Options {
701                log_level: Some(LogLevel::INFO),
702                ..Default::default()
703            }),
704        );
705
706        assert!(result.error.is_none());
707        assert_eq!(result.exit_code, 0);
708    }
709
710    /// Regression test for a deadlock where a buffered-but-unread output
711    /// channel filled and blocked the child process forever. Ported from
712    /// `TestRun_LargeOutputDoesNotDeadlock`.
713    ///
714    /// NOTE: `go-cmd`'s internal channel-capacity deadlock has no direct
715    /// Rust analogue when using `tokio::process::Command` with piped
716    /// stdout/stderr -- `tokio::io::AsyncRead` on a pipe does not have a
717    /// fixed internal line-buffer capacity the way `go-cmd`'s `Streaming`
718    /// channels do. But the *general class* of bug (spawn a child with
719    /// piped stdout/stderr, then don't drain one of the pipes promptly)
720    /// still deadlocks in Rust if the reader task isn't spawned before
721    /// `wait()`ing, because the OS pipe buffer itself is finite. This test
722    /// should be kept as a regression guard once ported to whatever
723    /// polling/streaming strategy Phase 4 picks. See PORT_NOTES.md.
724    #[test]
725    fn run_large_output_does_not_deadlock() {
726        let result = run(
727            "sh",
728            &[
729                "-c".into(),
730                "for i in $(seq 1 5000); do echo line $i; done".into(),
731            ],
732            None,
733        );
734
735        assert!(result.error.is_none(), "command should succeed");
736        assert_eq!(result.exit_code, 0, "exit code should be 0");
737        assert_eq!(
738            result.stdout.len(),
739            5000,
740            "all output lines should be captured"
741        );
742    }
743
744    #[test]
745    fn run_with_options_disable_streaming() {
746        let result = run(
747            "echo",
748            &["test".into()],
749            Some(Options {
750                disable_streaming: true,
751                ..Default::default()
752            }),
753        );
754
755        assert!(result.error.is_none());
756        assert_eq!(result.exit_code, 0);
757        assert!(!result.stdout.is_empty());
758    }
759
760    #[test]
761    fn look_path_finds_and_rejects() {
762        let path = look_path("sh").expect("sh should be found");
763        assert!(!path.as_os_str().is_empty());
764
765        assert!(look_path("nonexistent-command-xyz").is_err());
766    }
767
768    #[test]
769    fn result_combined_output() {
770        let result = ExecResult {
771            stdout: vec!["line1".into(), "line2".into()],
772            stderr: vec!["error1".into(), "error2".into()],
773            ..Default::default()
774        };
775
776        let output = result.combined_output();
777        assert!(output.contains("line1"));
778        assert!(output.contains("line2"));
779        assert!(output.contains("error1"));
780        assert!(output.contains("error2"));
781    }
782
783    #[test]
784    fn read_stdout_success() {
785        let stdout = read_stdout("echo", &["hello".into(), "world".into()]).unwrap();
786        assert!(stdout.contains("hello world"));
787    }
788
789    #[test]
790    fn read_stdout_multiple_lines() {
791        let stdout = read_stdout("sh", &["-c".into(), "echo line1; echo line2".into()]).unwrap();
792        assert!(stdout.contains("line1"));
793        assert!(stdout.contains("line2"));
794    }
795
796    #[test]
797    fn read_stdout_non_zero_exit() {
798        let err = read_stdout("sh", &["-c".into(), "exit 1".into()]).unwrap_err();
799        assert!(matches!(err, ExecError::CommandFailed(1)));
800    }
801
802    #[test]
803    fn read_stdout_empty_output() {
804        let stdout = read_stdout("sh", &["-c".into(), "".into()]).unwrap();
805        assert!(stdout.is_empty());
806    }
807
808    #[test]
809    fn run_non_zero_exit_default_failure_log_level() {
810        // Ported from TestRun_NonZeroExit_DefaultFailureLogLevel. Go
811        // captures a *slog.Logger writing to a buffer via
812        // `Options.Logger`; this port's equivalent is `Options.log_writer`
813        // (see its doc comment for why this isn't a full
814        // `tracing::Dispatch`).
815        let buf: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
816
817        let result = run(
818            "sh",
819            &["-c".into(), "exit 1".into()],
820            Some(Options {
821                log_writer: Some(buf.clone()),
822                ..Default::default()
823            }),
824        );
825
826        assert!(result.error.is_some());
827        assert_eq!(result.exit_code, 1);
828
829        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
830        assert!(output.contains("ERROR"));
831        assert!(output.contains("command error"));
832    }
833
834    #[test]
835    fn run_non_zero_exit_failure_log_level_override() {
836        let buf: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
837
838        let result = run(
839            "sh",
840            &["-c".into(), "exit 1".into()],
841            Some(Options {
842                failure_log_level: Some(LogLevel::DEBUG),
843                log_writer: Some(buf.clone()),
844                ..Default::default()
845            }),
846        );
847
848        assert!(result.error.is_some());
849        assert_eq!(result.exit_code, 1);
850
851        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
852        assert!(!output.contains("ERROR"));
853        assert!(output.contains("DEBUG"));
854        assert!(output.contains("command error"));
855    }
856
857    #[test]
858    fn check_success() {
859        let ok = check("true", &[]).unwrap();
860        assert!(ok);
861    }
862
863    #[test]
864    fn check_non_zero() {
865        let ok = check("false", &[]).unwrap();
866        assert!(!ok);
867    }
868
869    #[test]
870    fn check_binary_missing() {
871        let err = check("/no/such/binary-caproni-test", &[]).unwrap_err();
872        assert!(matches!(err, ExecError::RunFailed(_)));
873    }
874
875    #[test]
876    fn check_context_cancelled() {
877        // Go: context.WithTimeout(50ms). See `run_context_cancellation`'s
878        // comment -- this port threads the deadline explicitly since there
879        // is no ambient context to carry it, via `check_with_timeout`.
880        let err = check_with_timeout("sleep", &["10".into()], Some(StdDuration::from_millis(50)))
881            .unwrap_err();
882        assert!(matches!(err, ExecError::Cancelled));
883    }
884
885    #[test]
886    fn read_stdout_non_zero_exit_quiet_completion() {
887        let buf: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
888
889        let result = run(
890            "sh",
891            &["-c".into(), "echo x; exit 1".into()],
892            Some(Options {
893                log_level: Some(LogLevel::DEBUG),
894                stderr_log_level: Some(LogLevel::DEBUG),
895                failure_log_level: Some(LogLevel::DEBUG),
896                disable_streaming: true,
897                log_writer: Some(buf.clone()),
898                ..Default::default()
899            }),
900        );
901
902        assert!(matches!(result.error, Some(ExecError::CommandFailed(1))));
903        assert_eq!(result.stdout, vec!["x".to_string()]);
904
905        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
906        assert!(!output.contains("ERROR"));
907        assert!(output.contains("command error"));
908    }
909
910    struct SudoTestCase {
911        desc: &'static str,
912        in_name: &'static str,
913        in_args: &'static [&'static str],
914        out_name: &'static str,
915        out_args: &'static [&'static str],
916        uid: u32,
917        sudo_on_path: bool,
918    }
919
920    #[test]
921    fn apply_sudo_cases() {
922        let cases = [
923            SudoTestCase {
924                desc: "already root",
925                in_name: "update-ca-certificates",
926                in_args: &[],
927                out_name: "update-ca-certificates",
928                out_args: &[],
929                uid: 0,
930                sudo_on_path: false,
931            },
932            SudoTestCase {
933                desc: "sudo requested, available",
934                in_name: "security",
935                in_args: &["add-trusted-cert", "-d"],
936                out_name: "/usr/bin/sudo",
937                out_args: &["security", "add-trusted-cert", "-d"],
938                uid: 501,
939                sudo_on_path: true,
940            },
941            SudoTestCase {
942                desc: "sudo requested, not available",
943                in_name: "update-ca-certificates",
944                in_args: &[],
945                out_name: "update-ca-certificates",
946                out_args: &[],
947                uid: 501,
948                sudo_on_path: false,
949            },
950        ];
951
952        for case in cases {
953            let (name, args) = apply_sudo_with(
954                case.in_name,
955                &case
956                    .in_args
957                    .iter()
958                    .map(|s| s.to_string())
959                    .collect::<Vec<_>>(),
960                move || case.uid,
961                move |file| {
962                    if case.sudo_on_path {
963                        Ok(PathBuf::from(format!("/usr/bin/{file}")))
964                    } else {
965                        Err(ExecError::LookPathFailed(file.to_string()))
966                    }
967                },
968            );
969
970            assert_eq!(name, case.out_name, "{}", case.desc);
971            assert_eq!(
972                args,
973                case.out_args
974                    .iter()
975                    .map(|s| s.to_string())
976                    .collect::<Vec<_>>(),
977                "{}",
978                case.desc
979            );
980        }
981    }
982
983    #[test]
984    fn run_sudo_already_root() {
985        // Go stubs `geteuid`/`lookPath` package-level vars via
986        // `stubSudoEnv(t, 0, false)` for determinism (no real root
987        // privileges needed in CI). This port's equivalent is
988        // `run_sudo_with`'s explicit euid/lookup parameters -- see
989        // `apply_sudo_with`'s doc comment.
990        let err = run_sudo_with(
991            "true",
992            &[],
993            || 0,
994            |file| Err(ExecError::LookPathFailed(file.to_string())),
995        );
996        assert!(err.is_ok());
997    }
998
999    #[test]
1000    fn run_sudo_non_zero_exit() {
1001        let err = run_sudo_with(
1002            "sh",
1003            &["-c".into(), "exit 7".into()],
1004            || 0,
1005            |file| Err(ExecError::LookPathFailed(file.to_string())),
1006        )
1007        .unwrap_err();
1008        assert!(matches!(err, ExecError::CommandFailed(7)));
1009    }
1010}
xianyun_executor/exec/mod.rs

xianyun_executor/exec/
mod.rs
