sausage/host/tests/process.rs

//! Process acceptance tests use persistent userland build artifacts and temporary worlds.
use sausage::{
    exec::{Command, Exec, Options, Stream},
    process::Runtime,
    world::World,
};
use std::{
    path::{Path, PathBuf},
    sync::{Arc, Once},
    time::Duration,
};
static BUILD: Once = Once::new();
// Share the persistent compile cache without concurrent cold compilation of identical components.
static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
fn repo() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR"))
        .parent()
        .unwrap()
        .to_owned()
}
fn build() {
    BUILD.call_once(|| {
        let root = repo();
        let run = |args: &[&str], cwd: &Path| {
            assert!(
                std::process::Command::new(args[0])
                    .args(&args[1..])
                    .env_remove("CARGO_TARGET_DIR")
                    .env("CARGO_HOME", cwd.join(".cargo-home"))
                    .current_dir(cwd)
                    .status()
                    .unwrap()
                    .success(),
                "guest build failed: {args:?}"
            );
        };
        run(
            &["nu", "--no-config-file", "userland/cwd-init/build.nu"],
            &root,
        );
        run(
            &[
                "cargo",
                "build",
                "--release",
                "-p",
                "sausage-coreutils",
                "-p",
                "sausage-procps",
                "-p",
                "sausage-cwd-probe",
                "--locked",
            ],
            &root.join("userland"),
        );
        run(
            &["nu", "--no-config-file", "userland/brush/build.nu"],
            &root,
        );
    });
}
fn fixture(root: &Path) {
    for dir in ["work/sub", "other", "bin", "dev"] {
        std::fs::create_dir_all(root.join(dir)).unwrap();
    }
    for (name, text) in [
        ("work/f", "alpha\n"),
        ("work/file", "one\ntwo\n"),
        ("work/log", "before\n"),
        ("work/a.txt", ""),
        ("work/b.txt", ""),
        ("dev/null", ""),
        ("f", "root\n"),
    ] {
        std::fs::write(root.join(name), text).unwrap();
    }
}
fn world() -> (tempfile::TempDir, Arc<Runtime>, sausage::process::Processes) {
    build();
    let temp = tempfile::tempdir().unwrap();
    let world = World::open(temp.path()).unwrap();
    let cache = Path::new(env!("CARGO_TARGET_TMPDIR")).join("process-host/compile-cache");
    std::fs::create_dir_all(&cache).unwrap();
    std::os::unix::fs::symlink(cache, world.host_dir.join("compile-cache")).unwrap();
    let processes = world.processes.clone();
    fixture(&temp.path().join("world"));
    let artifacts = repo().join("userland/target/wasm32-wasip2/release");
    let bin = temp.path().join("world/bin");
    for tool in ["cat", "tr", "ls", "wc", "yes", "head", "sleep"] {
        std::fs::copy(artifacts.join("sausage-coreutils.wasm"), bin.join(tool)).unwrap();
    }
    for tool in ["ps", "kill"] {
        std::fs::copy(artifacts.join("sausage-procps.wasm"), bin.join(tool)).unwrap();
    }
    std::fs::copy(artifacts.join("sausage-cwd-probe.wasm"), bin.join("probe")).unwrap();
    std::fs::copy(
        repo().join("userland/brush/source/target/wasm32-wasip2/release/brush.wasm"),
        bin.join("sh"),
    )
    .unwrap();
    (temp, Runtime::new(world).unwrap(), processes)
}
async fn run(runtime: &Arc<Runtime>, command: Command, cwd: &str) -> (i32, Vec<u8>, Vec<u8>) {
    let (mut stdout, mut stderr) = (Vec::new(), Vec::new());
    let code = runtime
        .exec(
            command,
            Options {
                cwd: cwd.into(),
                timeout: Some(Duration::from_secs(30)),
                ..Default::default()
            },
            &mut |stream, bytes| match stream {
                Stream::Stdout => stdout.extend_from_slice(bytes),
                Stream::Stderr => stderr.extend_from_slice(bytes),
            },
        )
        .await
        .unwrap();
    (code, stdout, stderr)
}
async fn warm(runtime: &Arc<Runtime>) {
    // A cold compile belongs to setup, not to the command deadline or its timing.
    assert_eq!(
        runtime
            .exec(
                Command::Shell("sleep 0".into()),
                Options {
                    cwd: "/".into(),
                    ..Default::default()
                },
                &mut |_, _| {},
            )
            .await
            .unwrap(),
        0
    );
}
#[tokio::test(flavor = "multi_thread")]
async fn cwd_without_program_workaround() {
    let _serial = SERIAL.lock().await;
    let (_temp, runtime, _) = world();
    let (code, output, _) = run(&runtime, Command::Argv(vec!["/bin/probe".into()]), "/work").await;
    assert_eq!(code, 0);
    assert_eq!(output, b"cwd=/work\nrelative=alpha\nabsolute=root\n");
    for code in [0, 1, 2, 42, 255] {
        assert_eq!(
            run(
                &runtime,
                Command::Argv(vec!["/bin/probe".into(), code.to_string()]),
                "/"
            )
            .await
            .0,
            code
        );
    }
    let code = runtime
        .exec(
            Command::Argv(vec!["/bin/probe".into()]),
            Options {
                cwd: "/missing".into(),
                ..Default::default()
            },
            &mut |_, _| {},
        )
        .await
        .unwrap();
    assert_eq!(code, 1);
}
#[tokio::test(flavor = "multi_thread")]
async fn corpus_matches_bash() {
    let _serial = SERIAL.lock().await;
    let (temp, runtime, _) = world();
    warm(&runtime).await;
    let native = tempfile::tempdir().unwrap();
    fixture(native.path());
    let script = std::fs::read_to_string(repo().join("spikes/s2-shell/corpus.sh")).unwrap();
    let result = std::process::Command::new("bash")
        .args(["--noprofile", "--norc", "-c", &script])
        .current_dir(native.path())
        .output()
        .unwrap();
    let expected = String::from_utf8(result.stdout)
        .unwrap()
        .replace(native.path().to_str().unwrap(), "");
    let start = std::time::Instant::now();
    let (code, stdout, _) = run(&runtime, Command::Shell(script.clone()), "/").await;
    eprintln!("corpus elapsed={:?}", start.elapsed());
    assert_eq!(Some(code), result.status.code());
    assert_eq!(stdout, expected.as_bytes());
    fixture(&temp.path().join("world"));
    let cli = std::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
        .arg("sh")
        .arg(temp.path())
        .args(["-c", &script])
        .output()
        .unwrap();
    assert_eq!(cli.status.code(), result.status.code());
    assert_eq!(
        cli.stdout,
        expected.as_bytes(),
        "CLI stderr: {}",
        String::from_utf8_lossy(&cli.stderr)
    );
    for script in ["cat /work/f", "cd /work; cat f", "yes | head -n 2"] {
        let start = std::time::Instant::now();
        assert_eq!(run(&runtime, Command::Shell(script.into()), "/").await.0, 0);
        eprintln!("warm {script:?}: {:?}", start.elapsed());
    }
}
async fn empty(processes: &sausage::process::Processes) {
    tokio::time::timeout(Duration::from_secs(5), async {
        while !processes.lock().unwrap().is_empty() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap_or_else(|e| panic!("{e}: remaining processes {:?}", processes.lock().unwrap()));
}
#[tokio::test(flavor = "multi_thread")]
async fn timeout_and_drop_kill_descendants() {
    let _serial = SERIAL.lock().await;
    let (_temp, runtime, processes) = world();
    warm(&runtime).await;
    let result = runtime
        .exec(
            Command::Shell("sleep 60 | cat".into()),
            Options {
                timeout: Some(Duration::from_millis(100)),
                ..Default::default()
            },
            &mut |_, _| {},
        )
        .await;
    assert!(result.unwrap_err().to_string().contains("timed out"));
    empty(&processes).await;
    let task_runtime = runtime.clone();
    let task = tokio::spawn(async move {
        task_runtime
            .exec(
                Command::Shell("sleep 60 | cat".into()),
                Options::default(),
                &mut |_, _| {},
            )
            .await
    });
    tokio::time::timeout(Duration::from_secs(10), async {
        while processes.lock().unwrap().len() < 3 {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    task.abort();
    let _ = task.await;
    empty(&processes).await;
}
#[tokio::test(flavor = "multi_thread")]
async fn ps_and_kill_pipeline() {
    let _serial = SERIAL.lock().await;
    let (_temp, runtime, processes) = world();
    warm(&runtime).await;
    let task_runtime = runtime.clone();
    let task = tokio::spawn(async move {
        run(&task_runtime, Command::Shell("sleep 60 | cat".into()), "/").await
    });
    let pid = tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            if let Some((&pid, _)) = processes
                .lock()
                .unwrap()
                .iter()
                .find(|(_, info)| info.argv.first().is_some_and(|arg| arg.ends_with("sleep")))
            {
                break pid;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let (_, stdout, _) = run(&runtime, Command::Argv(vec!["/bin/ps".into()]), "/").await;
    assert!(
        String::from_utf8(stdout)
            .unwrap()
            .contains(&format!("{pid}\t"))
    );
    assert_eq!(
        run(
            &runtime,
            Command::Argv(vec!["/bin/kill".into(), pid.to_string()]),
            "/"
        )
        .await
        .0,
        0
    );
    tokio::time::timeout(Duration::from_secs(5), task)
        .await
        .unwrap()
        .unwrap();
    empty(&processes).await;
}
#[tokio::test(flavor = "multi_thread")]
async fn output_keeps_stream_order() {
    let _serial = SERIAL.lock().await;
    let (_temp, runtime, _) = world();
    warm(&runtime).await;
    let (code, stdout, stderr) = run(
        &runtime,
        Command::Shell("printf a; printf b; printf c >&2; printf d >&2".into()),
        "/",
    )
    .await;
    assert_eq!(code, 0);
    assert_eq!(stdout, b"ab");
    assert_eq!(stderr, b"cd");
}