vamp/tests/unit/shell.rs

use super::*;
use std::{
    fs,
    path::Path,
    sync::atomic::{AtomicU64, Ordering},
    time::Instant,
};

static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);

struct Fixture(PathBuf);

impl Fixture {
    fn new() -> Self {
        loop {
            let root = std::env::temp_dir().join(format!(
                "vamp-shell-{}-{}",
                std::process::id(),
                NEXT_ROOT.fetch_add(1, Ordering::Relaxed)
            ));
            match fs::create_dir(&root) {
                Ok(()) => return Self(fs::canonicalize(root).unwrap()),
                Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
                Err(error) => panic!("Cannot create shell fixture: {error}"),
            }
        }
    }
}

impl Drop for Fixture {
    fn drop(&mut self) {
        let _ = fs::remove_dir_all(&self.0);
    }
}

fn start(command: &str, root: &Path) -> Worker {
    Worker::start_with_shell(command.into(), root.into(), "/bin/sh".into()).unwrap()
}

fn wait_for<T>(mut ready: impl FnMut() -> Option<T>) -> T {
    // File gates determine ordering; this deadline only bounds a stuck test.
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        if let Some(value) = ready() {
            return value;
        }
        assert!(Instant::now() < deadline, "Shell fixture did not progress");
        thread::sleep(Duration::from_millis(1));
    }
}

fn process_running(pid: &str) -> bool {
    let output = Command::new("/bin/ps")
        .args(["-o", "stat=", "-p", pid])
        .output()
        .unwrap();
    let state = String::from_utf8(output.stdout).unwrap();
    let state = state.trim();
    // An orphan may await the system reaper, but a zombie cannot keep running
    // or retain the inherited output descriptors.
    !state.is_empty() && !state.starts_with('Z')
}

#[test]
fn shell_is_pending_until_gate_and_preserves_cwd_streams_status_and_null_stdin() {
    let fixture = Fixture::new();
    let mut worker = start(
        "printf '%s\\n' \"$PWD\"; : > ready; \
         while [ ! -e release ]; do :; done; \
         if read value; then exit 99; fi; \
         printf 'out\\000bytes'; printf 'err\\000bytes' >&2; exit 7",
        &fixture.0,
    );
    wait_for(|| fixture.0.join("ready").exists().then_some(()));
    assert!(worker.poll().is_none());
    fs::write(fixture.0.join("release"), b"").unwrap();
    let output = wait_for(|| worker.poll()).unwrap();
    let mut expected = fixture.0.as_os_str().as_encoded_bytes().to_vec();
    expected.extend_from_slice(b"\nout\0bytes");
    assert_eq!(output.stdout, expected);
    assert_eq!(output.stderr, b"err\0bytes");
    assert_eq!(output.code, Some(7));
    assert!(worker.poll().is_none());
}

#[test]
fn shell_reports_spawn_failure() {
    let fixture = Fixture::new();
    let mut worker = start("exit 0", &fixture.0.join("missing"));
    let error = wait_for(|| worker.poll()).unwrap_err();
    assert!(error.starts_with("Cannot start shell command:"), "{error}");
}

#[test]
fn disconnected_shell_worker_reports_one_failure() {
    let (cancel, _cancelled) = mpsc::channel();
    let (send, results) = mpsc::channel();
    drop(send);
    let mut worker = Worker {
        cancel,
        results,
        thread: None,
        finished: false,
    };
    assert_eq!(
        worker.poll().unwrap().unwrap_err(),
        "Shell worker stopped unexpectedly"
    );
    assert!(worker.poll().is_none());
}

#[test]
fn shell_drains_both_streams_concurrently_with_explicit_bounded_capture() {
    let fixture = Fixture::new();
    let mut worker = start(
        "(dd if=/dev/zero bs=1024 count=4097 2>/dev/null) & \
         (dd if=/dev/zero bs=1024 count=4097 2>/dev/null) >&2 & wait",
        &fixture.0,
    );
    let output = wait_for(|| worker.poll()).unwrap();
    assert_eq!(output.code, Some(0));
    for bytes in [output.stdout, output.stderr] {
        assert_eq!(bytes.len(), OUTPUT_LIMIT + TRUNCATION_NOTE.len());
        assert!(bytes[..OUTPUT_LIMIT].iter().all(|byte| *byte == 0));
        assert_eq!(&bytes[OUTPUT_LIMIT..], TRUNCATION_NOTE);
    }
}

#[test]
fn dropping_shell_kills_and_reaps_group_with_pipe_inheriting_descendant() {
    let fixture = Fixture::new();
    let worker = start(
        "/bin/sh -c 'printf \"%s\" \"$$\" > descendant; : > ready; \
         while [ ! -e release ]; do :; done' & \
         printf '%s' \"$$\" > leader; wait",
        &fixture.0,
    );
    wait_for(|| fixture.0.join("ready").exists().then_some(()));
    let descendant = fs::read_to_string(fixture.0.join("descendant")).unwrap();
    let leader = wait_for(|| {
        fs::read_to_string(fixture.0.join("leader"))
            .ok()
            .filter(|pid| !pid.is_empty())
    });
    let (done, dropped) = mpsc::channel();
    let dropper = thread::spawn(move || {
        drop(worker);
        done.send(()).unwrap();
    });
    dropped.recv_timeout(Duration::from_secs(10)).unwrap();
    dropper.join().unwrap();
    wait_for(|| (!process_running(&descendant)).then_some(()));
    let leader_status = Command::new("/bin/ps")
        .args(["-o", "stat=", "-p", &leader])
        .output()
        .unwrap();
    assert!(
        leader_status.stdout.is_empty(),
        "Shell leader was not reaped"
    );
}

#[test]
fn completed_shell_does_not_leave_background_jobs_holding_output_open() {
    let fixture = Fixture::new();
    let mut worker = start(
        "/bin/sh -c 'printf \"%s\" \"$$\" > descendant; : > ready; \
         while [ ! -e release ]; do :; done' & \
         while [ ! -e ready ]; do :; done; printf done",
        &fixture.0,
    );
    let output = wait_for(|| worker.poll()).unwrap();
    assert_eq!(output.code, Some(0));
    assert_eq!(output.stdout, b"done");
    let descendant = fs::read_to_string(fixture.0.join("descendant")).unwrap();
    wait_for(|| (!process_running(&descendant)).then_some(()));
}