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,
};
mod features {
    //! Acceptance coverage for preview1, identity, inherited IO, services, HTTP, and bundle aliases.
    use super::*;
    use std::{collections::BTreeMap, time::Instant};

    async fn poll(mut condition: impl FnMut() -> bool) {
        tokio::time::timeout(Duration::from_secs(10), async {
            while !condition() {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
    }
    fn lines(path: &Path) -> Vec<String> {
        std::fs::read_to_string(path)
            .unwrap_or_default()
            .lines()
            .map(str::to_owned)
            .collect()
    }
    async fn prepare(runtime: &Arc<Runtime>, argv: Vec<String>) {
        assert_eq!(
            runtime
                .exec(
                    Command::Argv(argv),
                    Options {
                        cwd: "/".into(),
                        ..Default::default()
                    },
                    &mut |_, _| {}
                )
                .await
                .unwrap(),
            0
        );
    }
    #[tokio::test(flavor = "multi_thread")]
    async fn core_module_exit_and_cancellation() {
        let _serial = SERIAL.lock().await;
        let (temp, runtime, processes) = world();
        let bin = temp.path().join("world/bin");
        std::fs::write(bin.join("exit23"), wat::parse_str(r#"(module (import "wasi_snapshot_preview1" "proc_exit" (func (param i32))) (memory (export "memory") 1) (func (export "_start") i32.const 23 call 0))"#).unwrap()).unwrap();
        assert_eq!(
            run(&runtime, Command::Argv(vec!["/bin/exit23".into()]), "/")
                .await
                .0,
            23
        );
        std::fs::write(
            bin.join("spin"),
            wat::parse_str(
                r#"(module (memory (export "memory") 1) (func (export "_start") (loop br 0)))"#,
            )
            .unwrap(),
        )
        .unwrap();
        let result = runtime
            .exec(
                Command::Argv(vec!["/bin/spin".into()]),
                Options {
                    timeout: Some(Duration::from_millis(100)),
                    ..Default::default()
                },
                &mut |_, _| {},
            )
            .await;
        assert!(result.unwrap_err().to_string().contains("timed out"));
        empty(&processes).await;
    }
    #[tokio::test(flavor = "multi_thread")]
    async fn preview1_pipeline_and_esbuild() {
        let _serial = SERIAL.lock().await;
        let (temp, runtime, _) = world();
        warm(&runtime).await;
        let root = temp.path().join("world");
        std::fs::copy("/tmp/wasi-userland-lab/qjs-wasi.wasm", root.join("bin/qjs")).unwrap();
        std::fs::copy("/tmp/s6-esbuild/esbuild.wasm", root.join("bin/esbuild")).unwrap();
        prepare(&runtime, vec!["/bin/qjs".into(), "--help".into()]).await;
        let direct = run(
            &runtime,
            Command::Argv(vec![
                "/bin/qjs".into(),
                "--std".into(),
                "-e".into(),
                "console.log(42)".into(),
            ]),
            "/",
        )
        .await;
        assert_eq!(direct.0, 0, "{}", String::from_utf8_lossy(&direct.2));
        assert_eq!(direct.1, b"42\n");
        let (code, out, err) = run(
        &runtime,
        Command::Shell(
            "echo 'console.log(6*7)' | qjs --std -e 'std.evalScript(std.in.readAsString())' | cat"
                .into(),
        ),
        "/",
    )
    .await;
        assert_eq!(code, 0, "{}", String::from_utf8_lossy(&err));
        assert_eq!(out, b"42\n");
        assert_eq!(
            run(
                &runtime,
                Command::Shell("qjs --std -e 'std.exit(7)'".into()),
                "/"
            )
            .await
            .0,
            7
        );
        for name in ["app.tsx", "package.json", "package-lock.json"] {
            std::fs::copy(repo().join("spikes/s6-esbuild").join(name), root.join(name)).unwrap();
        }
        assert!(
            std::process::Command::new("cp")
                .args(["-r", "/tmp/s6-esbuild/app/node_modules"])
                .arg(&root)
                .status()
                .unwrap()
                .success()
        );
        prepare(&runtime, vec!["/bin/esbuild".into(), "--version".into()]).await;
        let flags = [
            "app.tsx",
            "--bundle",
            "--minify",
            "--format=iife",
            "--platform=browser",
            "--jsx=automatic",
            "--jsx-import-source=preact",
            "--legal-comments=none",
            "--log-level=error",
        ];
        assert!(
            std::process::Command::new("/tmp/s6-esbuild/esbuild-native")
                .current_dir(&root)
                .args(flags)
                .arg("--outfile=native.js")
                .status()
                .unwrap()
                .success()
        );
        let script = format!("esbuild {} --outfile=module.js", flags.join(" "));
        let (code, _, err) = run(&runtime, Command::Shell(script), "/").await;
        assert_eq!(code, 0, "{}", String::from_utf8_lossy(&err));
        assert_eq!(
            std::fs::read(root.join("native.js")).unwrap(),
            std::fs::read(root.join("module.js")).unwrap()
        );
    }
    #[tokio::test(flavor = "multi_thread")]
    async fn inherited_stdio_identity_and_conversation() {
        let _serial = SERIAL.lock().await;
        let (_temp, runtime, processes) = world();
        warm(&runtime).await;
        let (code, out, err) = run(
            &runtime,
            Command::Shell(
                "echo inherited | /bin/launcher inherit /bin/launcher read | cat".into(),
            ),
            "/",
        )
        .await;
        assert_eq!(code, 0, "{}", String::from_utf8_lossy(&err));
        assert_eq!(out, b"inherited\n");
        let (code, out, err) = run(
            &runtime,
            Command::Shell("/bin/launcher inherit /bin/launcher emit".into()),
            "/",
        )
        .await;
        assert_eq!(code, 0);
        assert_eq!(out, b"out\n");
        assert_eq!(err, b"err\n");
        let (code, out, err) = run(
            &runtime,
            Command::Shell("echo unused | /bin/launcher null /bin/launcher check-null".into()),
            "/",
        )
        .await;
        assert_eq!(code, 0);
        assert!(out.is_empty());
        assert!(err.is_empty());
        assert_eq!(
            run(
                &runtime,
                Command::Argv(vec!["/bin/launcher".into(), "path".into()]),
                "/"
            )
            .await
            .1,
            b"/usr/local/bin:/bin\n"
        );
        let mut out = Vec::new();
        let code = runtime
            .exec(
                Command::Argv(vec![
                    "/bin/launcher".into(),
                    "identity".into(),
                    "/bin/launcher".into(),
                ]),
                Options {
                    cwd: "/".into(),
                    env: vec![("SAUSAGE_PID".into(), "spoof".into())],
                    ..Default::default()
                },
                &mut |_, bytes| out.extend_from_slice(bytes),
            )
            .await
            .unwrap();
        assert_eq!(code, 0);
        let text = String::from_utf8(out).unwrap();
        let ids: BTreeMap<_, _> = text
            .lines()
            .map(|line| {
                let (name, value) = line.split_once('=').unwrap();
                (name, value.parse::<u32>().unwrap())
            })
            .collect();
        assert_eq!(ids["pid"], ids["child"]);
        assert_ne!(ids["pid"], ids["parent"]);
        let exec = runtime.clone();
        let task = tokio::spawn(async move {
            exec.exec(
                Command::Shell("sleep 60 | cat".into()),
                Options {
                    conversation: Some("c1".into()),
                    ..Default::default()
                },
                &mut |_, _| {},
            )
            .await
        });
        poll(|| processes.lock().unwrap().len() >= 3).await;
        assert!(
            processes
                .lock()
                .unwrap()
                .values()
                .all(|p| p.conversation.as_deref() == Some("c1"))
        );
        let (_, out, _) = run(&runtime, Command::Shell("cat /proc/processes".into()), "/").await;
        assert!(String::from_utf8(out).unwrap().matches("\tc1\t").count() >= 3);
        task.abort();
        let _ = task.await;
        empty(&processes).await;
    }
    #[tokio::test(flavor = "multi_thread")]
    async fn relative_aliases_share_compilation() {
        let _serial = SERIAL.lock().await;
        let (temp, runtime, _) = world();
        let bin = temp.path().join("world/bin");
        for name in ["true", "echo"] {
            std::os::unix::fs::symlink("cat", bin.join(name)).unwrap();
        }
        assert_eq!(
            run(&runtime, Command::Argv(vec!["/bin/true".into()]), "/")
                .await
                .0,
            0
        );
        assert_eq!(runtime.compiled_programs(), 1);
        let (code, out, _) = run(
            &runtime,
            Command::Argv(vec!["/bin/echo".into(), "shared".into()]),
            "/",
        )
        .await;
        assert_eq!(code, 0);
        assert_eq!(out, b"shared\n");
        assert_eq!(runtime.compiled_programs(), 1);
    }
    #[tokio::test(flavor = "multi_thread")]
    async fn services_restart_edit_limit_and_eof() {
        let _serial = SERIAL.lock().await;
        let (temp, runtime, processes) = world();
        prepare(&runtime, vec!["/bin/launcher".into()]).await;
        let root = temp.path().join("world");
        std::fs::create_dir_all(root.join("config")).unwrap();
        let spec = |name: &str, mode: &str, log: &str| serde_json::json!({"name": name, "argv": ["/bin/launcher", mode, log]});
        let write = |specs: Vec<serde_json::Value>| {
            std::fs::write(
                root.join("config/services.yaml"),
                serde_yaml_ng::to_string(&specs).unwrap(),
            )
            .unwrap()
        };
        write(vec![
            spec("a", "server", "/a.log"),
            spec("crash", "crash", "/crash.log"),
        ]);
        let supervisor = sausage::process::supervise(runtime.clone());
        poll(|| {
            lines(&root.join("a.log")) == ["start"] && lines(&root.join("crash.log")).len() == 6
        })
        .await;
        prepare(&runtime, vec!["/bin/launcher".into()]).await;
        assert_eq!(lines(&root.join("crash.log")).len(), 6);
        assert!(
            processes
                .lock()
                .unwrap()
                .values()
                .all(|p| p.conversation.is_none())
        );
        write(vec![spec("b", "server", "/b.log")]);
        poll(|| {
            lines(&root.join("a.log")) == ["start", "stop"]
                && lines(&root.join("b.log")) == ["start"]
        })
        .await;
        let mut changed = spec("b", "server", "/b.log");
        changed["env"] = serde_json::json!({"REV": "new"});
        write(vec![changed]);
        poll(|| lines(&root.join("b.log")) == ["start", "stop", "start"]).await;
        write(vec![]);
        poll(|| lines(&root.join("b.log")) == ["start", "stop", "start", "stop"]).await;
        write(vec![spec("linger", "linger", "/linger.log")]);
        poll(|| lines(&root.join("linger.log")) == ["start"]).await;
        let stop = Instant::now();
        write(vec![]);
        poll(|| processes.lock().unwrap().is_empty()).await;
        assert!(stop.elapsed() >= Duration::from_secs(5));
        supervisor.abort();
        let _ = supervisor.await;
        empty(&processes).await;
    }
    #[tokio::test(flavor = "multi_thread")]
    async fn git_local_and_anonymous_https_clone() {
        let _serial = SERIAL.lock().await;
        let (temp, runtime, _) = world();
        warm(&runtime).await;
        prepare(&runtime, vec!["/bin/git".into(), "--help".into()]).await;
        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = runtime.exec(Command::Shell("git init /repo && cd /repo && echo hello > file && git add file && git commit -m 'process test' && git log --oneline".into()),
        Options { cwd: "/".into(), env: vec![("GIT_AUTHOR_NAME".into(), "Test".into()), ("GIT_AUTHOR_EMAIL".into(), "test@example.com".into()), ("GIT_COMMITTER_NAME".into(), "Test".into()), ("GIT_COMMITTER_EMAIL".into(), "test@example.com".into())], ..Default::default() },
        &mut |stream, bytes| match stream { Stream::Stdout => out.extend_from_slice(bytes), Stream::Stderr => err.extend_from_slice(bytes) }).await.unwrap();
        assert_eq!(code, 0, "{}", String::from_utf8_lossy(&err));
        assert!(String::from_utf8(out).unwrap().contains("process test"));
        let (code, _, err) = run(
            &runtime,
            Command::Shell("git clone https://github.com/octocat/Hello-World.git /clone".into()),
            "/",
        )
        .await;
        assert_eq!(code, 0, "{}", String::from_utf8_lossy(&err));
        assert!(
            std::fs::read_to_string(temp.path().join("world/clone/README"))
                .unwrap()
                .contains("Hello World")
        );
    }
}
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,
        );
        run(
            &[
                "cargo",
                "build",
                "--release",
                "--target",
                "wasm32-wasip2",
                "--manifest-path",
                "brush/probe/Cargo.toml",
                "--locked",
            ],
            &root.join("userland"),
        );
        run(
            &[
                "cargo",
                "build",
                "--release",
                "-p",
                "git",
                "--target-dir",
                "../target/process-userland",
                "--locked",
            ],
            &root.join("userland"),
        );
    });
}
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();
    std::fs::copy(
        repo().join("userland/brush/probe/target/wasm32-wasip2/release/process-probe.wasm"),
        bin.join("launcher"),
    )
    .unwrap();
    std::fs::copy(
        repo().join("target/process-userland/wasm32-wasip2/release/git.wasm"),
        bin.join("git"),
    )
    .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");
}