sausage/cli/tests/core.rs

//! End-to-end host behavior with a raw WASI guest.
use sausage::{Host, world::World};
use serde_json::{Value, json};
use std::{path::PathBuf, sync::OnceLock};
fn fixture() -> PathBuf {
    static WASM: OnceLock<PathBuf> = OnceLock::new();
    WASM.get_or_init(|| {
        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .parent()
            .unwrap()
            .to_path_buf();
        let target = root.join("target/core/guests");
        let status = std::process::Command::new("cargo")
            .args([
                "build",
                "--offline",
                "--target",
                "wasm32-wasip2",
                "--manifest-path",
            ])
            .arg(root.join("cli/tests/plugins/probe/Cargo.toml"))
            .arg("--target-dir")
            .arg(&target)
            .status()
            .unwrap();
        assert!(status.success());
        target.join("wasm32-wasip2/debug/probe.wasm")
    })
    .clone()
}
struct TestWorld(PathBuf);
impl TestWorld {
    fn new() -> Self {
        let p = std::env::temp_dir().join(format!("sausage-{}", uuid::Uuid::now_v7()));
        let w = World::create(&p).unwrap();
        std::fs::create_dir_all(w.plugin_dir("probe")).unwrap();
        std::fs::copy(fixture(), w.plugin_dir("probe").join("plugin.wasm")).unwrap();
        Self(p)
    }
    async fn start(&self) -> Host {
        Host::start(&self.0).await.unwrap()
    }
}
impl Drop for TestWorld {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}
async fn call(h: &Host, m: &str, p: Value) -> Value {
    tokio::time::timeout(std::time::Duration::from_secs(20), h.call(m, p))
        .await
        .unwrap()
        .unwrap()
}
#[tokio::test(flavor = "multi_thread")]
async fn routing_environment_files_and_errors() {
    let w = TestWorld::new();
    let h = w.start().await;
    assert_eq!(call(&h, "probe.echo", json!({"x":1})).await, json!({"x":1}));
    assert_eq!(
        call(&h, "probe.rpc.discover", json!({"x":2})).await,
        json!({"x":2})
    );
    assert_eq!(
        call(
            &h,
            "probe.call",
            json!({"method":"probe.echo","params":{"nested":true}})
        )
        .await,
        json!({"nested":true})
    );
    assert_eq!(
        call(&h, "probe.multicall", json!({})).await,
        json!({"count":2})
    );
    call(
        &h,
        "probe.notification",
        json!({"path":"/state/notification","text":"delivered"}),
    )
    .await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/notification")).unwrap(),
        "delivered"
    );
    assert_eq!(
        h.call("probe.unknownid", json!({})).await.unwrap_err()["code"],
        -32002
    );
    let env = call(&h, "probe.env", json!({})).await;
    assert_eq!(env["plugin"], "probe");
    assert_eq!(env["parent"], "console");
    assert!(env["pid"].as_str().unwrap().starts_with("probe-"));
    call(
        &h,
        "probe.write",
        json!({"path":"/state/test","text":"hello"}),
    )
    .await;
    assert_eq!(
        call(&h, "probe.read", json!({"path":"/state/test"})).await,
        "hello"
    );
    assert_eq!(
        h.call("absent.run", json!({})).await.unwrap_err()["code"],
        -32001
    );
    assert_eq!(
        h.call("core.activate", json!({})).await.unwrap_err()["code"],
        -32601
    );
    assert_eq!(
        h.call("probe.echo", json!([])).await.unwrap_err()["code"],
        -32602
    );
    let e = h.call("probe.exit", json!({})).await.unwrap_err();
    assert_eq!(e["code"], -32002);
    assert!(
        e["data"]["lines"]
            .as_array()
            .unwrap()
            .iter()
            .any(|l| l["line"] == "last line")
    );
    h.stop().await;
}
#[tokio::test(flavor = "multi_thread")]
async fn journal_notifications_locks_and_kill() {
    let w = TestWorld::new();
    let h = w.start().await;
    let mut notifications = h.subscribe();
    call(&h, "probe.notify", json!({})).await;
    assert_eq!(
        notifications.recv().await.unwrap()["params"]["name"],
        "progress"
    );
    call(&h, "probe.junk", json!({})).await;
    tokio::time::timeout(std::time::Duration::from_secs(5), async {
        while !h.processes().is_empty() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let starts = call(&h, "core.log", json!({"grep":"start junk"})).await;
    let pid = &starts[0]["pid"];
    let warnings = call(&h, "core.log", json!({"pid":pid,"grep":"discarded stdout"})).await;
    assert_eq!(warnings.as_array().unwrap().len(), 1);
    assert!(!warnings.to_string().contains("junk"));
    let log = call(&h, "core.log", json!({"grep":"stderr line"})).await;
    assert_eq!(log.as_array().unwrap().len(), 1);
    assert_eq!(
        call(&h, "core.lock", json!({"path":"x"})).await,
        json!({"held":true})
    );
    assert_eq!(
        call(&h, "core.lock", json!({"path":"x","wait":false})).await,
        json!({"held":false})
    );
    call(&h, "core.unlock", json!({"path":"x"})).await;
    assert_eq!(
        call(&h, "probe.lock", json!({"path":"x"})).await,
        json!({"held":true})
    );
    let pid = call(&h, "probe.serve", json!({})).await;
    call(&h, "core.kill", json!({"pid":pid})).await;
    assert!(!h.processes().contains_key(pid.as_str().unwrap()));
    h.stop().await;
}
#[tokio::test(flavor = "multi_thread")]
async fn native_and_reserved_world_paths() {
    let w = TestWorld::new();
    let h = w.start().await;
    h.register_native(
        "native",
        std::sync::Arc::new(|ctx, method, params| {
            Box::pin(async move {
                ctx.notify("native", json!({}));
                Ok(json!({"pid":ctx.pid,"method":method,"params":params}))
            })
        }),
    );
    let result = call(&h, "native.echo", json!({"x":1})).await;
    assert_eq!(result["method"], "echo");
    assert!(result["pid"].as_str().unwrap().starts_with("native-"));
    h.stop().await;
    for name in ["proc", "run", "self"] {
        std::fs::create_dir(w.0.join(name)).unwrap();
        assert!(World::open(&w.0).is_err());
        std::fs::remove_dir(w.0.join(name)).unwrap();
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn tree_detach_stop_boot_and_scheduler() {
    let w = TestWorld::new();
    std::fs::write(
        w.0.join("config/boot.json"),
        serde_json::to_vec(&json!([
            {"method":"absent.start","params":{}},
            {"method":"probe.write","params":{"path":"/state/boot","text":"ready"}},
            {"method":"probe.read","params":{"path":"/state/boot"}}
        ]))
        .unwrap(),
    )
    .unwrap();
    let h = w.start().await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/boot")).unwrap(),
        "ready"
    );
    let attached = call(
        &h,
        "probe.call",
        json!({"method":"probe.serve","params":{}}),
    )
    .await;
    tokio::time::timeout(std::time::Duration::from_secs(7), async {
        while h.processes().contains_key(attached.as_str().unwrap()) {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let detached = call(
        &h,
        "probe.call",
        json!({"method":"core.call","params":{"method":"probe.serve","params":{},"detach":true}}),
    )
    .await;
    assert!(h.processes()[detached.as_str().unwrap()].detached);
    call(&h, "core.stop", json!({"pid":detached})).await;
    let hc = h.clone();
    let busy = tokio::spawn(async move { hc.call("probe.busy", json!({})).await });
    let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
        loop {
            if let Some((pid, _)) = h.processes().into_iter().find(|(_, p)| p.method == "busy") {
                break pid;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert_eq!(
        call(&h, "probe.echo", json!({"works":true})).await,
        json!({"works":true})
    );
    call(&h, "core.kill", json!({"pid":pid})).await;
    assert_eq!(busy.await.unwrap().unwrap_err()["code"], -32002);
    h.stop().await;
}

#[tokio::test(flavor = "multi_thread")]
async fn same_process_lock_acquisitions_queue() {
    let w = TestWorld::new();
    let h = w.start().await;
    h.register_native(
        "locks",
        std::sync::Arc::new(|ctx, _, _| {
            Box::pin(async move {
                assert_eq!(
                    ctx.call("core.lock", json!({"path":"fifo"})).await.unwrap(),
                    json!({"held":true})
                );
                assert_eq!(
                    ctx.call("core.lock", json!({"path":"fifo","wait":false}))
                        .await
                        .unwrap(),
                    json!({"held":false})
                );
                let mut second = Box::pin(ctx.call("core.lock", json!({"path":"fifo"})));
                std::future::poll_fn(|cx| {
                    assert!(second.as_mut().poll(cx).is_pending());
                    std::task::Poll::Ready(())
                })
                .await;
                ctx.call("core.unlock", json!({"path":"fifo"}))
                    .await
                    .unwrap();
                assert_eq!(second.await.unwrap(), json!({"held":true}));
                Ok(json!({}))
            })
        }),
    );
    call(&h, "locks.test", json!({})).await;
    h.stop().await;
    assert_eq!(
        call(&h, "core.lock", json!({"path":"fifo","wait":false})).await,
        json!({"held":true})
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn stop_closes_stdin_deadline_and_exit() {
    let w = TestWorld::new();
    let h = w.start().await;
    let hc = h.clone();
    let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
    let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
        loop {
            if let Some((pid, _)) = h
                .processes()
                .into_iter()
                .find(|(_, p)| p.method == "wait_eof")
            {
                break pid;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    call(&h, "core.stop", json!({"pid":pid})).await;
    assert_eq!(
        std::fs::read_to_string(w.0.join("state/stopped")).unwrap(),
        "yes"
    );
    assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);
    assert_eq!(call(&h, "core.exit", json!({})).await, json!({}));
    tokio::time::timeout(std::time::Duration::from_secs(1), h.exit_requested())
        .await
        .unwrap();
    h.stop().await;
}

#[tokio::test(flavor = "multi_thread")]
async fn tcp_udp_and_dns() {
    use tokio::io::AsyncWriteExt;
    let w = TestWorld::new();
    let h = w.start().await;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let tcp = format!("localhost:{}", listener.local_addr().unwrap().port());
    let udp = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
    let udp_addr = udp.local_addr().unwrap().to_string();
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        stream.write_all(b"tcp").await.unwrap();
    });
    assert_eq!(
        call(&h, "probe.network", json!({"tcp":tcp,"udp":udp_addr})).await,
        "tcp"
    );
    let mut buf = [0; 3];
    udp.recv_from(&mut buf).await.unwrap();
    assert_eq!(&buf, b"udp");
    server.await.unwrap();
    h.stop().await;
}

#[tokio::test(flavor = "multi_thread")]
async fn native_kill_drops_future_and_releases_locks() {
    use std::sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    };
    struct Dropped(Arc<AtomicBool>);
    impl Drop for Dropped {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }
    let w = TestWorld::new();
    let h = w.start().await;
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = dropped.clone();
    h.register_native(
        "native",
        Arc::new(move |ctx, _, _| {
            let flag = flag.clone();
            Box::pin(async move {
                let _guard = Dropped(flag);
                ctx.call("core.lock", json!({"path":"native-lock"}))
                    .await
                    .unwrap();
                ctx.notify("ready", json!({"pid":ctx.pid}));
                std::future::pending().await
            })
        }),
    );
    let mut notifications = h.subscribe();
    let hc = h.clone();
    let pending = tokio::spawn(async move { hc.call("native.wait", json!({})).await });
    let notification =
        tokio::time::timeout(std::time::Duration::from_secs(10), notifications.recv())
            .await
            .unwrap()
            .unwrap();
    let pid = &notification["params"]["params"]["pid"];
    assert_eq!(
        call(&h, "core.lock", json!({"path":"native-lock","wait":false})).await,
        json!({"held":false})
    );
    call(&h, "core.kill", json!({"pid":pid})).await;
    tokio::time::timeout(std::time::Duration::from_secs(1), async {
        while !dropped.load(Ordering::SeqCst) {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert_eq!(
        call(&h, "core.lock", json!({"path":"native-lock","wait":false})).await,
        json!({"held":true})
    );
    assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);
    h.stop().await;
}

#[tokio::test(flavor = "multi_thread")]
async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
    let w = TestWorld::new();
    let h = w.start().await;
    let mut tasks = Vec::new();
    for _ in 0..2 {
        let hc = h.clone();
        tasks.push(tokio::spawn(async move {
            hc.call("probe.busy", json!({})).await
        }));
    }
    tokio::time::timeout(std::time::Duration::from_secs(20), async {
        while h.processes().len() != 2 {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    let before = std::time::Instant::now();
    h.stop().await;
    assert!(before.elapsed() >= std::time::Duration::from_millis(4900));
    assert!(before.elapsed() < std::time::Duration::from_secs(7));
    for task in tasks {
        assert_eq!(task.await.unwrap().unwrap_err()["code"], -32002);
    }
    assert!(h.processes().is_empty());
}

#[tokio::test(flavor = "multi_thread")]
async fn journal_filters_persist_and_process_versions_are_hashes() {
    use sha2::{Digest, Sha256};
    let w = TestWorld::new();
    let h = w.start().await;
    let pid = call(&h, "probe.serve", json!({"version":1})).await;
    let pid = pid.as_str().unwrap();
    let info = h.processes()[pid].clone();
    assert_eq!(
        info.version,
        format!("{:x}", Sha256::digest(std::fs::read(fixture()).unwrap()))
    );
    assert_eq!(info.plugin, "probe");
    assert_eq!(info.method, "serve");
    assert_eq!(info.params, json!({"version":1}));
    assert_eq!(info.parent.as_deref(), Some("console"));
    assert_eq!(info.state, "running");
    assert!(!info.detached);
    assert!(info.start_time > 0);
    let uuid = uuid::Uuid::parse_str(pid.strip_prefix("probe-").unwrap()).unwrap();
    assert_eq!(uuid.get_version_num(), 7);
    call(&h, "core.stop", json!({"pid":pid})).await;
    let entries = call(
        &h,
        "core.log",
        json!({"plugin":"probe","pid":pid,"limit":2}),
    )
    .await;
    assert_eq!(entries.as_array().unwrap().len(), 2);
    let entry = &entries[0];
    let id = &entry["id"];
    let single = call(&h, "core.log", json!({"since":id,"until":id})).await;
    assert_eq!(single, json!([entry]));
    h.stop().await;
    drop(h);
    let h = w.start().await;
    assert_eq!(
        call(&h, "core.log", json!({"since":id,"until":id})).await,
        single
    );
    h.stop().await;
    assert!(
        std::fs::read(w.0.join("state/core/journal.sqlite"))
            .unwrap()
            .starts_with(b"SQLite format 3")
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn stopping_rejects_calls_but_allows_notifications() {
    use std::sync::{Arc, Mutex};
    let w = TestWorld::new();
    let h = w.start().await;
    let (tx, rx) = tokio::sync::oneshot::channel();
    let tx = Arc::new(Mutex::new(Some(tx)));
    let gate = Arc::new(tokio::sync::Notify::new());
    let release = gate.clone();
    h.register_native(
        "native",
        Arc::new(move |ctx, _, _| {
            let tx = tx.lock().unwrap().take().unwrap();
            let gate = gate.clone();
            Box::pin(async move {
                tx.send(ctx.clone()).ok();
                gate.notified().await;
                Ok(json!({}))
            })
        }),
    );
    let hc = h.clone();
    let pending = tokio::spawn(async move { hc.call("native.wait", json!({})).await });
    let ctx = rx.await.unwrap();
    let hc = h.clone();
    let pid = ctx.pid.clone();
    let stop = tokio::spawn(async move { hc.call("core.stop", json!({"pid":pid})).await });
    tokio::time::timeout(std::time::Duration::from_secs(2), async {
        while h.processes()[&ctx.pid].state != "stopping" {
            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap();
    assert_eq!(
        ctx.call("probe.echo", json!({})).await.unwrap_err()["code"],
        -32600
    );
    let mut notifications = h.subscribe();
    ctx.call("core.notify", json!({"name":"stop-hook","params":{}}))
        .await
        .unwrap();
    assert_eq!(
        notifications.recv().await.unwrap()["params"]["name"],
        "stop-hook"
    );
    release.notify_one();
    assert_eq!(pending.await.unwrap().unwrap(), json!({}));
    assert_eq!(stop.await.unwrap().unwrap(), json!({}));
    h.stop().await;
}