sausage/cli/tests/common/mod.rs

use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use sausage::{Envelope, Host, World};
use serde_json::{Value, json};
use tokio::sync::mpsc;

const REPLY_TIMEOUT: Duration = Duration::from_secs(10);
static WORLD_SEQUENCE: AtomicU64 = AtomicU64::new(0);

pub fn repo_root() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .parent()
        .unwrap()
        .to_path_buf()
}

pub fn create_world(label: &str) -> World {
    let nonce = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    let sequence = WORLD_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    let path = repo_root()
        .join("target/test-worlds")
        .join(format!("{label}-{}-{nonce}-{sequence}", std::process::id()));
    World::create(&path).unwrap()
}

pub fn actor_fixture_source(name: &str) -> PathBuf {
    repo_root().join("cli/tests/actors").join(name)
}

pub fn cargo_source(name: &str) -> PathBuf {
    repo_root().join("cli/tests/crates").join(name)
}

pub fn toolbox_wasm() -> &'static PathBuf {
    static WASM: OnceLock<PathBuf> = OnceLock::new();
    WASM.get_or_init(|| build_fixture("toolbox"))
}

fn build_fixture(name: &str) -> PathBuf {
    let root = repo_root();
    let manifest = actor_fixture_source(name).join("Cargo.toml");
    let target_dir = root.join("target/test-actors");
    let output = Command::new("cargo")
        .args([
            "build",
            "--release",
            "--target",
            "wasm32-wasip2",
            "--target-dir",
        ])
        .arg(&target_dir)
        .arg("--manifest-path")
        .arg(&manifest)
        .current_dir(&root)
        .output()
        .expect("run cargo to build the test actor");
    assert!(
        output.status.success(),
        "building fixture {name} failed\nstdout:\n{}\nstderr:\n{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    let wasm = target_dir
        .join("wasm32-wasip2/release")
        .join(format!("{name}.wasm"));
    assert!(
        wasm.is_file(),
        "fixture build did not create {}",
        wasm.display()
    );
    wasm
}

pub fn passing_source(world: &World, name: &str) -> PathBuf {
    source_with_test(
        world,
        name,
        json!({
            "steps": [{
                "send": {"type": "echo", "value": "candidate"},
                "expect": {"type": "echo", "value": "candidate"}
            }]
        }),
    )
}

pub fn source_with_test(world: &World, name: &str, test: Value) -> PathBuf {
    let source = world.source_dir("console").join("test-fixtures").join(name);
    std::fs::create_dir_all(source.join("tests")).unwrap();
    std::fs::write(
        source.join("tests/candidate.json"),
        serde_json::to_vec(&test).unwrap(),
    )
    .unwrap();
    source
}

pub async fn start(world: &World) -> (Host, mpsc::Receiver<Envelope>) {
    let host = Host::start(&world.root).await.unwrap();
    let mailbox = host.mailbox("console");
    (host, mailbox)
}

pub async fn request(host: &Host, mailbox: &mut mpsc::Receiver<Envelope>, body: Value) -> Envelope {
    let id = host
        .broker
        .send("console", Some("broker"), None, None, body);
    receive_reply(mailbox, &id, REPLY_TIMEOUT).await
}

pub async fn receive_message(
    mailbox: &mut mpsc::Receiver<Envelope>,
    timeout: Duration,
) -> Envelope {
    tokio::time::timeout(timeout, mailbox.recv())
        .await
        .expect("timed out waiting for a message")
        .expect("console mailbox closed")
}

pub async fn receive_reply(
    mailbox: &mut mpsc::Receiver<Envelope>,
    id: &str,
    timeout: Duration,
) -> Envelope {
    tokio::time::timeout(timeout, async {
        loop {
            let envelope = mailbox.recv().await.expect("console mailbox closed");
            if envelope.reply_to.as_deref() == Some(id) {
                return envelope;
            }
        }
    })
    .await
    .unwrap_or_else(|_| panic!("timed out waiting for reply to {id}"))
}

pub async fn receive_replies(
    mailbox: &mut mpsc::Receiver<Envelope>,
    ids: &[String],
    timeout: Duration,
) -> Vec<Envelope> {
    let mut replies = Vec::with_capacity(ids.len());
    tokio::time::timeout(timeout, async {
        while replies.len() < ids.len() {
            let message = mailbox.recv().await.expect("console mailbox closed");
            assert!(
                ids.iter()
                    .any(|id| message.reply_to.as_deref() == Some(id.as_str())),
                "unexpected message while waiting for replies: {message:?}"
            );
            replies.push(message);
        }
    })
    .await
    .expect("timed out waiting for replies");
    replies
}

pub async fn ask(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
    actor: &str,
    body: Value,
) -> Envelope {
    let id = host.broker.send("console", Some(actor), None, None, body);
    receive_reply(mailbox, &id, REPLY_TIMEOUT).await
}

pub async fn ask_type(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
    actor: &str,
    body: Value,
    expected: &str,
) -> Envelope {
    let reply = ask(host, mailbox, actor, body).await;
    assert_eq!(reply.body["type"], expected, "unexpected reply: {reply:?}");
    reply
}

pub fn spawn_request(host: &Host, actor: &str, wasm: &Path, source: &Path) -> String {
    host.broker.send(
        "console",
        Some("broker"),
        None,
        None,
        json!({
            "type": "spawn",
            "actor": actor,
            "wasm": wasm.display().to_string(),
            "source": source.display().to_string()
        }),
    )
}

pub async fn spawn_actor(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
    actor: &str,
    wasm: &Path,
    source: &Path,
) -> Envelope {
    let id = spawn_request(host, actor, wasm, source);
    receive_reply(mailbox, &id, REPLY_TIMEOUT).await
}

pub async fn spawn_type(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
    actor: &str,
    wasm: &Path,
    source: &Path,
) -> Envelope {
    let reply = spawn_actor(host, mailbox, actor, wasm, source).await;
    assert_eq!(
        reply.body["type"], "spawned",
        "unexpected spawn reply: {reply:?}"
    );
    reply
}

pub async fn upgrade_actor(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
    actor: &str,
    wasm: &Path,
    source: &Path,
) -> Envelope {
    request(
        host,
        mailbox,
        json!({
            "type": "upgrade",
            "actor": actor,
            "wasm": wasm.display().to_string(),
            "source": source.display().to_string()
        }),
    )
    .await
}