400 lines
14 KiB
Rust
400 lines
14 KiB
Rust
//! Exercise SDK plugins through the real host, without a fake core.
|
|
use sausage::{Host, world::World};
|
|
use serde_json::{Value, json};
|
|
use std::{path::PathBuf, sync::OnceLock, time::Duration};
|
|
|
|
fn fixtures() -> &'static PathBuf {
|
|
static TARGET: OnceLock<PathBuf> = OnceLock::new();
|
|
TARGET.get_or_init(|| {
|
|
for (package, binary) in [
|
|
("examples/notes", "notes"),
|
|
("examples/consumer", "consumer"),
|
|
("examples/www", "www"),
|
|
("examples/secrets", "secrets"),
|
|
("sdk/tests/plugins/consumer", "sdk-consumer"),
|
|
] {
|
|
super::common::build(package, binary);
|
|
}
|
|
super::common::sdk_probe().parent().unwrap().to_owned()
|
|
})
|
|
}
|
|
struct TestWorld(PathBuf);
|
|
impl TestWorld {
|
|
fn new() -> Self {
|
|
let path = std::env::temp_dir().join(format!("sausage-sdk-{}", uuid::Uuid::now_v7()));
|
|
let world = World::create(&path).unwrap();
|
|
for (plugin, binary) in [
|
|
("notes", "notes"),
|
|
("consumer", "consumer"),
|
|
("www", "www"),
|
|
("secrets", "secrets"),
|
|
("probe", "sdk-probe"),
|
|
("typed", "sdk-consumer"),
|
|
] {
|
|
std::fs::create_dir_all(world.plugin_dir(plugin)).unwrap();
|
|
std::fs::copy(
|
|
fixtures().join(format!("{binary}.wasm")),
|
|
world.plugin_dir(plugin).join("plugin.wasm"),
|
|
)
|
|
.unwrap();
|
|
}
|
|
Self(path)
|
|
}
|
|
async fn start(&self) -> Host {
|
|
Host::start_with_cache(
|
|
&self.0,
|
|
std::path::Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache")),
|
|
)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
}
|
|
impl Drop for TestWorld {
|
|
fn drop(&mut self) {
|
|
let _ = std::fs::remove_dir_all(&self.0);
|
|
}
|
|
}
|
|
async fn call(host: &Host, method: &str, params: Value) -> Value {
|
|
tokio::time::timeout(Duration::from_secs(60), host.call(method, params))
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
}
|
|
|
|
/// Concurrent read-modify-write calls retain every note under the SDK lock.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn notes_lock_preserves_parallel_additions() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "notes.add", json!({"text":"first"})).await["count"],
|
|
1
|
|
);
|
|
let calls = (0..12)
|
|
.map(|n| {
|
|
let h = host.clone();
|
|
tokio::spawn(
|
|
async move { call(&h, "notes.add", json!({"text":format!("note-{n}")})).await },
|
|
)
|
|
})
|
|
.collect::<Vec<_>>();
|
|
for task in calls {
|
|
task.await.unwrap();
|
|
}
|
|
let notes = call(&host, "notes.list", json!({})).await;
|
|
assert_eq!(notes.as_array().unwrap().len(), 13);
|
|
assert_eq!(notes[0], "first");
|
|
for n in 0..12 {
|
|
assert!(
|
|
notes
|
|
.as_array()
|
|
.unwrap()
|
|
.contains(&json!(format!("note-{n}")))
|
|
);
|
|
}
|
|
host.stop().await;
|
|
}
|
|
/// The example consumer invokes notes through its generated import.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn generated_notes_import() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "consumer.note", json!({"text":"typed"})).await,
|
|
1
|
|
);
|
|
assert_eq!(call(&host, "notes.list", json!({})).await, json!(["typed"]));
|
|
host.stop().await;
|
|
}
|
|
/// Discovery retains the complete document and satisfies the consumer contract.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn discovery_is_compatible() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let actual = call(&host, "notes.rpc.discover", json!({})).await;
|
|
let consumer = call(&host, "consumer.rpc.discover", json!({})).await;
|
|
let expected: Value = serde_json::from_str(include_str!(
|
|
"../../../examples/consumer/imports/notes.json"
|
|
))
|
|
.unwrap();
|
|
assert_eq!(actual, expected);
|
|
assert_eq!(actual["openrpc"], "1.2.6");
|
|
assert!(actual["components"]["schemas"]["AddParams"].is_object());
|
|
assert!(
|
|
actual["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.find(|m| m["name"] == "list")
|
|
.unwrap()
|
|
.get("x-sausage-params")
|
|
.is_none()
|
|
);
|
|
assert!(actual["components"]["schemas"].get("Empty").is_none());
|
|
assert!(
|
|
sausage_interface::compat::check(&consumer["x-sausage-imports"]["notes"], &actual)
|
|
.is_empty()
|
|
);
|
|
host.stop().await;
|
|
}
|
|
/// Detached SDK calls receive progress and forward it through core.notify.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn detached_call_notifications() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let mut events = host.subscribe();
|
|
assert_eq!(call(&host, "probe.relay", json!({})).await, 42);
|
|
let event = tokio::time::timeout(Duration::from_secs(5), events.recv())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(event["params"]["name"], "step");
|
|
host.stop().await;
|
|
}
|
|
/// Detached typed calls complete and detached services survive their original caller.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn detached_service_survives_parent() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(call(&host, "typed.run", json!({})).await, 5);
|
|
let child = call(&host, "probe.launch", json!({})).await;
|
|
tokio::time::timeout(Duration::from_secs(5), async {
|
|
while host.processes().values().any(|p| p.method == "launch") {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert!(host.processes().contains_key(child.as_str().unwrap()));
|
|
host.stop().await;
|
|
}
|
|
/// A core.stop request closes stdin and executes the SDK stop hook.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn stop_runs_hook() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let pid = call(&host, "probe.service", json!({})).await;
|
|
call(&host, "core.stop", json!({"pid":pid})).await;
|
|
assert_eq!(std::fs::read(world.0.join("stopped")).unwrap(), b"yes");
|
|
host.stop().await;
|
|
}
|
|
/// PROTOCOL 3.7: an idle SDK listener opts into drain and exits before EOF/grace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn superseded_idle_listener_drains_before_grace() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let (notice, mut notices) = tokio::sync::mpsc::unbounded_channel();
|
|
host.register_native(
|
|
"other",
|
|
std::sync::Arc::new(move |_, method, _| {
|
|
let notice = notice.clone();
|
|
Box::pin(async move {
|
|
assert_eq!(method, "superseded");
|
|
notice.send(std::time::Instant::now()).unwrap();
|
|
Ok(json!({}))
|
|
})
|
|
}),
|
|
);
|
|
let pid = call(&host, "probe.drain_service", json!({})).await;
|
|
// Replace with a different real component; force only skips unrelated test consumers.
|
|
std::fs::copy(
|
|
fixtures().join("notes.wasm"),
|
|
world.0.join("state/next.wasm"),
|
|
)
|
|
.unwrap();
|
|
call(
|
|
&host,
|
|
"core.activate",
|
|
json!({"plugin":"probe","wasm":"/state/next.wasm","force":true}),
|
|
)
|
|
.await;
|
|
// Activation preparation/discovery is outside upgrade grace. Time from the
|
|
// hook's outgoing call, not from the start of candidate compilation.
|
|
let before = tokio::time::timeout(Duration::from_secs(60), notices.recv())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
// Waiting on its lifetime lock is an exit event, not process-liveness polling.
|
|
call(&host, "core.lock", json!({"path":"/drain-owner"})).await;
|
|
assert!(before.elapsed() < Duration::from_secs(5));
|
|
assert!(!host.processes().contains_key(pid.as_str().unwrap()));
|
|
assert!(
|
|
!world
|
|
.0
|
|
.join(format!("stopped-{}", pid.as_str().unwrap()))
|
|
.exists(),
|
|
"drain must not run on_stop (discovery processes may run their own hooks)"
|
|
);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// SDK framing carries payloads above 64 KiB in both directions.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn large_messages() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let text = "z".repeat(192 * 1024);
|
|
assert_eq!(call(&host, "probe.echo", json!({"text":text})).await, text);
|
|
host.stop().await;
|
|
}
|
|
/// Normal SDK stdout never produces the discarded-output journal warning.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn no_discarded_stdout() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
call(&host, "probe.echo", json!({"text":"framing"})).await;
|
|
host.stop().await;
|
|
assert_eq!(
|
|
call(&host, "core.log", json!({"grep":"discarded stdout"})).await,
|
|
json!([])
|
|
);
|
|
}
|
|
/// Two SDK processes exchange endpoint requests and progress at a world path.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn world_endpoint() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
call(&host, "probe.service", json!({})).await;
|
|
assert_eq!(call(&host, "typed.world_endpoint", json!({})).await, 7);
|
|
host.stop().await;
|
|
}
|
|
/// The HTTP example exposes its self endpoint through the process namespace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn www_control_through_proc() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "consumer.demo", json!({"text":"web","port":0})).await,
|
|
0
|
|
);
|
|
host.stop().await;
|
|
}
|
|
/// The vault example publishes a fixed endpoint in the shared run namespace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn secrets_fixed_endpoint() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
call(&host, "secrets.serve", json!({})).await;
|
|
assert_eq!(call(&host, "consumer.vault", json!({})).await, "hello");
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.6: a background task retains its SDK process until the task completes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn background_task_retains_process() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let pid = call(&host, "probe.background_hold", json!({})).await;
|
|
assert!(host.processes().contains_key(pid.as_str().unwrap()));
|
|
assert_eq!(call(&host, "notes.list", json!({})).await, json!([]));
|
|
assert!(host.processes().contains_key(pid.as_str().unwrap()));
|
|
std::fs::write(world.0.join("release-background"), "release").unwrap();
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
while host.processes().contains_key(pid.as_str().unwrap()) {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
std::fs::read_to_string(world.0.join("background-ended")).unwrap(),
|
|
"done"
|
|
);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 5: stale endpoint identities and interfaces fail through genuine SDK processes.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn endpoint_handshake_rejects_mismatch() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let pid = call(&host, "probe.service", json!({})).await;
|
|
let path = world.0.join("state/control");
|
|
let original = std::fs::read_to_string(&path).unwrap();
|
|
let words: Vec<_> = original.split_whitespace().collect();
|
|
for changed in [
|
|
format!("{} wrong-owner {}\n", words[0], words[2]),
|
|
format!("{} {} wrong-interface\n", words[0], words[1]),
|
|
] {
|
|
std::fs::write(&path, changed).unwrap();
|
|
let error = host
|
|
.call("typed.world_endpoint", json!({}))
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(error["code"], -32000);
|
|
assert!(error["message"].as_str().unwrap().contains("mismatch"));
|
|
assert!(host.processes().contains_key(pid.as_str().unwrap()));
|
|
}
|
|
std::fs::write(path, original).unwrap();
|
|
assert_eq!(call(&host, "typed.world_endpoint", json!({})).await, 7);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 10.1 and 5: discovery through core retains params, endpoints, imports, and grace.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn discovery_extensions_through_core() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
let provider = call(&host, "probe.rpc.discover", json!({})).await;
|
|
assert_eq!(provider["openrpc"], "1.2.6");
|
|
assert_eq!(provider["x-sausage-grace"], 12);
|
|
assert!(
|
|
provider["x-sausage-endpoints"]["control"]["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|m| m["name"] == "stats")
|
|
);
|
|
let add = provider["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.find(|m| m["name"] == "add")
|
|
.unwrap();
|
|
assert_eq!(add["paramStructure"], "by-name");
|
|
let reference = add["x-sausage-params"]["$ref"].as_str().unwrap();
|
|
assert_eq!(
|
|
provider.pointer(reference.trim_start_matches('#')).unwrap()["type"],
|
|
"object"
|
|
);
|
|
let consumer = call(&host, "typed.rpc.discover", json!({})).await;
|
|
assert!(
|
|
consumer["x-sausage-imports"]["probe"]["methods"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|m| m["name"] == "add")
|
|
);
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.2: rpc.discover is SDK control traffic and endpoint control is not a main method.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn reserved_rpc_control_names() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert!(call(&host, "probe.rpc.discover", json!({})).await["methods"].is_array());
|
|
for method in ["probe.rpc.endpoint", "probe.rpc.unknown"] {
|
|
assert_eq!(
|
|
host.call(method, json!({})).await.unwrap_err()["code"],
|
|
-32601
|
|
);
|
|
}
|
|
host.stop().await;
|
|
}
|
|
|
|
/// PROTOCOL 3.1: the SDK's leading newline isolates an unfinished diagnostic from its response.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn leading_newline_protects_sdk_response() {
|
|
let world = TestWorld::new();
|
|
let host = world.start().await;
|
|
assert_eq!(
|
|
call(&host, "probe.framed_output", json!({})).await,
|
|
"intact"
|
|
);
|
|
host.stop().await;
|
|
let warnings = call(&host, "core.log", json!({"grep":"discarded stdout"})).await;
|
|
assert_eq!(warnings.as_array().unwrap().len(), 1);
|
|
assert!(!warnings.to_string().contains("partial diagnostic"));
|
|
}
|