swale 0.1.0

A scheduled asset-graph orchestrator on object storage.
Documentation
use std::path::Path;
use std::process::Command;

use swale::{OperatorSet, Partitioning};

const EXAMPLE: &str = "examples/orders_daily.toml";

/// An empty directory under cargo's target tmp dir, left in place after the
/// test for inspection.
fn scratch(name: &str) -> std::path::PathBuf {
    let dir = Path::new(env!("CARGO_TARGET_TMPDIR")).join(name);
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    dir
}

#[test]
fn example_graph_loads_with_the_documented_structure() {
    let graph = swale::load_path(Path::new(EXAMPLE), &OperatorSet::builtin()).unwrap();
    assert_eq!(graph.name(), "orders_daily");
    assert_eq!(graph.schedule(), Some("0 2 * * *"));
    assert_eq!(
        graph.catchup(),
        Some(std::time::Duration::from_secs(7 * 86_400))
    );
    assert_eq!(graph.partitioning(), Partitioning::Daily);
    let order: Vec<&str> = graph.nodes().iter().map(|n| n.name()).collect();
    assert_eq!(
        order,
        ["extract", "transform", "load", "partition_ready", "notify"]
    );
    let roots: Vec<&str> = graph.roots().map(|n| n.name()).collect();
    assert_eq!(roots, ["extract", "partition_ready"]);
    let leaves: Vec<&str> = graph.leaves().map(|n| n.name()).collect();
    assert_eq!(leaves, ["partition_ready", "notify"]);
    assert_eq!(graph.edge_count(), 3);
    assert_eq!(graph.node("load").unwrap().pool(), "warehouse");
    assert_eq!(graph.node("load").unwrap().retries(), 5);
}

#[test]
fn validate_command_reports_a_valid_definition_with_status_0() {
    let output = Command::new(env!("CARGO_BIN_EXE_swale"))
        .args(["validate", EXAMPLE])
        .output()
        .unwrap();
    assert_eq!(output.status.code(), Some(0));
    assert_eq!(
        String::from_utf8(output.stdout).unwrap(),
        "orders_daily: 5 nodes, 3 edges\n"
    );
}

#[test]
fn validate_command_prints_each_fault_with_status_1() {
    let dir = scratch("swale-validate");
    let path = dir.join("bad.toml");
    std::fs::write(
        &path,
        r#"
[graph]
name = "g"
[[node]]
name = "a"
produces = "a"
consumes = ["missing"]
operator = "sql"
"#,
    )
    .unwrap();
    let output = Command::new(env!("CARGO_BIN_EXE_swale"))
        .args(["validate", path.to_str().unwrap()])
        .output()
        .unwrap();
    assert_eq!(output.status.code(), Some(1));
    assert_eq!(
        String::from_utf8(output.stderr).unwrap(),
        "error: node `a`: consumes `missing`, which no node produces\n\
         error: node `a`: unknown operator `sql`\n"
    );
}

#[test]
fn validate_command_reports_a_missing_file_with_status_1() {
    let output = Command::new(env!("CARGO_BIN_EXE_swale"))
        .args(["validate", "/nonexistent/graph.toml"])
        .output()
        .unwrap();
    assert_eq!(output.status.code(), Some(1));
    assert!(
        String::from_utf8(output.stderr)
            .unwrap()
            .starts_with("error: cannot read the definition: ")
    );
}

#[test]
fn usage_error_has_status_2() {
    let output = Command::new(env!("CARGO_BIN_EXE_swale")).output().unwrap();
    assert_eq!(output.status.code(), Some(2));
    assert!(
        String::from_utf8(output.stderr)
            .unwrap()
            .contains("Usage: swale")
    );
}

const LOCAL_EXAMPLE: &str = "examples/local.toml";

fn runnable() -> String {
    std::fs::read_to_string(LOCAL_EXAMPLE).unwrap()
}

fn run_command(
    dir: &std::path::Path,
    file: &std::path::Path,
    extra: &[&str],
) -> std::process::Output {
    Command::new(env!("CARGO_BIN_EXE_swale"))
        .arg("run")
        .arg(file)
        .arg("--store")
        .arg(dir.join("store"))
        .args(extra)
        .output()
        .unwrap()
}

#[test]
fn run_command_completes_a_graph_on_a_directory_store_and_resumes_an_existing_run() {
    let dir = scratch("swale-run");
    let file = dir.join("local.toml");
    std::fs::write(&file, runnable()).unwrap();

    let output = run_command(&dir, &file, &[]);
    let stdout = String::from_utf8(output.stdout).unwrap();
    assert_eq!(
        output.status.code(),
        Some(0),
        "{stdout}\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(
        stdout.starts_with("local/none: started, 1 root node(s) submitted\n"),
        "{stdout}"
    );
    assert!(
        stdout.contains("  first: succeeded (local-none-first-r0)\n"),
        "{stdout}"
    );
    assert!(
        stdout.contains("  second: succeeded (local-none-second-r0)\n"),
        "{stdout}"
    );
    assert!(stdout.ends_with("local/none: complete\n"), "{stdout}");

    let output = run_command(&dir, &file, &[]);
    let stdout = String::from_utf8(output.stdout).unwrap();
    assert_eq!(output.status.code(), Some(0), "{stdout}");
    assert!(
        stdout.starts_with("local/none: run exists, resuming\n"),
        "{stdout}"
    );
    assert!(stdout.ends_with("local/none: complete\n"), "{stdout}");
}

#[test]
fn run_command_accepts_a_file_url_and_refuses_an_unknown_or_unbuilt_scheme() {
    let dir = scratch("swale-run-url");
    let file = dir.join("local.toml");
    std::fs::write(&file, runnable()).unwrap();
    let store = dir.join("store");
    std::fs::create_dir_all(&store).unwrap();
    let url = format!("file://{}", store.display());
    let output = Command::new(env!("CARGO_BIN_EXE_swale"))
        .args(["run", file.to_str().unwrap(), "--store", &url])
        .output()
        .unwrap();
    let stdout = String::from_utf8(output.stdout).unwrap();
    assert_eq!(
        output.status.code(),
        Some(0),
        "{stdout}\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(stdout.ends_with("local/none: complete\n"), "{stdout}");

    for (store, message) in [
        ("ftp://bucket/prefix", "scheme `ftp` is not one of"),
        ("s3://bucket/prefix", "needs a build with the `aws` feature"),
    ] {
        let output = Command::new(env!("CARGO_BIN_EXE_swale"))
            .args(["run", file.to_str().unwrap(), "--store", store])
            .output()
            .unwrap();
        assert_eq!(output.status.code(), Some(2), "{store}");
        let stderr = String::from_utf8(output.stderr).unwrap();
        assert!(stderr.contains(message), "{store}: {stderr}");
    }
}

#[test]
fn run_command_reports_a_failed_run_with_status_1() {
    let dir = scratch("swale-run-fail");
    let file = dir.join("local.toml");
    std::fs::write(
        &file,
        runnable().replace("printf '{\\\"n\\\": 1}'", "echo no >&2; exit 4"),
    )
    .unwrap();

    let output = run_command(&dir, &file, &[]);
    let stdout = String::from_utf8(output.stdout).unwrap();
    assert_eq!(output.status.code(), Some(1), "{stdout}");
    assert!(
        stdout.contains("  first: failed (local-none-first-r0)\n    `sh` exited with 4: no\n"),
        "{stdout}"
    );
    assert!(stdout.ends_with("local/none: failed\n"), "{stdout}");
}

#[test]
fn run_command_requires_a_partition_for_a_partitioned_graph() {
    let dir = scratch("swale-run-part");
    let file = dir.join("daily.toml");
    std::fs::write(
        &file,
        runnable().replace(
            "name = \"local\"",
            "name = \"local\"\npartition = \"daily\"",
        ),
    )
    .unwrap();

    let output = run_command(&dir, &file, &[]);
    assert_eq!(output.status.code(), Some(1));
    assert_eq!(
        String::from_utf8(output.stderr).unwrap(),
        "error: the graph is partitioned: pass --partition <key>\n"
    );
    let output = run_command(&dir, &file, &["--partition", "20260915"]);
    let stdout = String::from_utf8(output.stdout).unwrap();
    assert_eq!(output.status.code(), Some(0), "{stdout}");
    assert!(stdout.contains("(local-20260915-first-r0)"), "{stdout}");
}