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 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 spawn_actor(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
    actor: &str,
    wasm: &Path,
    source: &Path,
) -> Envelope {
    request(
        host,
        mailbox,
        json!({
            "type": "spawn",
            "actor": actor,
            "wasm": wasm.display().to_string(),
            "source": source.display().to_string()
        }),
    )
    .await
}

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
}

pub async fn wait_for_parked(world: &World, actor: &str, message_id: &str) {
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let status_path = world.actor_dir(actor).join("status.json");
            let parked_id = std::fs::read_to_string(status_path)
                .ok()
                .and_then(|text| serde_json::from_str::<Value>(&text).ok())
                .and_then(|status| {
                    status
                        .pointer("/parked_message/id")
                        .and_then(Value::as_str)
                        .map(str::to_string)
                });
            if parked_id.as_deref() == Some(message_id) {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("actor did not park the crashed message");
}