ironflow-engine 2.40.2

//! Integration tests for artifact production and consumption.
//!
//! Black-box: real shell processes writing real files into a real temporary
//! directory, stored through the real `LocalBlobStore`.

use std::collections::HashMap;
use std::sync::Arc;

use serde_json::json;
use tempfile::TempDir;
use uuid::Uuid;

use ironflow_artifacts::blob_store::BlobStore;
use ironflow_artifacts::local::LocalBlobStore;
use ironflow_core::provider::AgentProvider;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_core::providers::record_replay::RecordReplayProvider;
use ironflow_engine::artifact::{ArtifactSink, DirectArtifactSink};
use ironflow_engine::config::{ArtifactInput, ShellConfig};
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::error::EngineError;
use ironflow_store::memory::InMemoryStore;
use ironflow_store::models::{NewRun, StepStatus, TriggerKind};
use ironflow_store::store::Store;

/// A run context wired to a filesystem-backed artifact store.
struct Fixture {
    ctx: WorkflowContext,
    store: Arc<dyn Store>,
    run_id: Uuid,
    /// Where steps run and write their files.
    work_dir: TempDir,
    /// Where artifact blobs are persisted.
    _blob_dir: TempDir,
}

impl Fixture {
    async fn new() -> Self {
        Self::build(true).await
    }

    /// Same fixture, without an artifact backend attached.
    async fn without_artifact_storage() -> Self {
        Self::build(false).await
    }

    async fn build(with_artifacts: bool) -> Self {
        let work_dir = TempDir::new().expect("work dir");
        let blob_dir = TempDir::new().expect("blob dir");

        let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
        let provider: Arc<dyn AgentProvider> = Arc::new(RecordReplayProvider::replay(
            ClaudeCodeProvider::new(),
            "/tmp/ironflow-fixtures",
        ));

        let run = store
            .create_run(NewRun {
                workflow_name: "artifacts".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                created_by: None,
                idempotency_key: None,
                max_cost_usd: None,
            })
            .await
            .expect("create run")
            .into_run();

        let mut ctx = WorkflowContext::new(run.id, "test".to_string(), store.clone(), provider);
        if with_artifacts {
            let blob: Arc<dyn BlobStore> = Arc::new(LocalBlobStore::new(blob_dir.path()));
            let sink: Arc<dyn ArtifactSink> =
                Arc::new(DirectArtifactSink::new(blob, store.clone()));
            ctx.set_artifact_sink(sink);
        }

        Self {
            ctx,
            store,
            run_id: run.id,
            work_dir,
            _blob_dir: blob_dir,
        }
    }

    /// A shell config rooted in the fixture's working directory.
    fn shell(&self, command: &str) -> ShellConfig {
        ShellConfig::new(command).dir(self.work_dir.path().to_str().expect("utf-8 path"))
    }
}

#[tokio::test]
async fn a_declared_output_is_stored_with_its_size_type_and_hash() {
    let mut fixture = Fixture::new().await;

    fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'hello' > report.html")
                .output("report.html"),
        )
        .await
        .expect("step");

    let artifacts = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list");

    assert_eq!(artifacts.len(), 1);
    assert_eq!(artifacts[0].name, "report.html");
    assert_eq!(artifacts[0].content_type, "text/html");
    assert_eq!(artifacts[0].size_bytes, 5);
    assert_eq!(
        artifacts[0].sha256,
        "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
    );
}

#[tokio::test]
async fn a_later_step_reads_an_artifact_produced_earlier() {
    let mut fixture = Fixture::new().await;

    let build = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'payload' > report.txt")
                .output("report.txt"),
        )
        .await
        .expect("producer");
    let report = build.artifact("report.txt").expect("declared by build");
    assert_eq!(report.step(), "build");
    assert_eq!(report.name(), "report.txt");

    // The consumer runs in a different directory: the only way it can see the
    // file is through the declared input.
    let consumer_dir = TempDir::new().expect("consumer dir");
    let consumer = ShellConfig::new("cat report.txt")
        .dir(consumer_dir.path().to_str().expect("utf-8 path"))
        .input(&report);

    let output = fixture
        .ctx
        .shell("publish", consumer)
        .await
        .expect("consumer");

    assert_eq!(output.stdout(), "payload");
}

#[tokio::test]
async fn an_input_can_be_written_to_another_path() {
    let mut fixture = Fixture::new().await;

    let build = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'x' > report.txt")
                .output("report.txt"),
        )
        .await
        .expect("producer");
    let report = build.artifact("report.txt").expect("declared by build");

    let consumer_dir = TempDir::new().expect("consumer dir");
    let consumer = ShellConfig::new("cat nested/renamed.txt")
        .dir(consumer_dir.path().to_str().expect("utf-8 path"))
        .input_at(&report, "nested/renamed.txt");

    let output = fixture
        .ctx
        .shell("publish", consumer)
        .await
        .expect("consumer");

    assert_eq!(output.stdout(), "x");
}

#[tokio::test]
async fn a_name_the_step_did_not_declare_is_refused() {
    let mut fixture = Fixture::new().await;

    let build = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'x' > report.txt")
                .output("report.txt"),
        )
        .await
        .expect("producer");

    // The typo is caught where the handle is taken, not in a later step.
    let err = build
        .artifact("report.html")
        .expect_err("report.html was never declared");

    assert!(matches!(
        err,
        EngineError::ArtifactNotDeclared { ref step, ref name }
            if step == "build" && name == "report.html"
    ));
}

#[tokio::test]
async fn a_step_without_declared_outputs_hands_out_no_artifact() {
    let mut fixture = Fixture::new().await;

    let greet = fixture
        .ctx
        .shell("greet", fixture.shell("echo hello"))
        .await
        .expect("step");

    assert!(matches!(
        greet.artifact("hello.txt"),
        Err(EngineError::ArtifactNotDeclared { .. })
    ));
}

#[tokio::test]
async fn a_glob_declaration_hands_out_the_files_it_matches() {
    let mut fixture = Fixture::new().await;

    let build = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'first' > a.log; touch notes.txt")
                .output("*.log"),
        )
        .await
        .expect("producer");

    let a_log = build.artifact("a.log").expect("matches *.log");
    assert!(matches!(
        build.artifact("notes.txt"),
        Err(EngineError::ArtifactNotDeclared { .. })
    ));

    let bytes = fixture.ctx.get_artifact(&a_log).await.expect("stored");
    assert_eq!(bytes, b"first");
}

#[tokio::test]
async fn an_artifact_is_named_after_the_file_not_the_declared_path() {
    let mut fixture = Fixture::new().await;

    let build = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("mkdir -p target && printf 'x' > target/report.html")
                .output("target/report.html"),
        )
        .await
        .expect("producer");

    assert!(build.artifact("report.html").is_ok());
    assert!(matches!(
        build.artifact("target/report.html"),
        Err(EngineError::ArtifactNotDeclared { .. })
    ));
}

#[tokio::test]
async fn a_glob_stores_every_match() {
    let mut fixture = Fixture::new().await;

    fixture
        .ctx
        .shell(
            "build",
            fixture.shell("touch a.log b.log notes.txt").output("*.log"),
        )
        .await
        .expect("step");

    let names: Vec<String> = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list")
        .into_iter()
        .map(|artifact| artifact.name)
        .collect();

    assert_eq!(names.len(), 2);
    assert!(names.contains(&"a.log".to_string()));
    assert!(names.contains(&"b.log".to_string()));
}

#[tokio::test]
async fn an_output_that_matches_nothing_fails_a_successful_step() {
    let mut fixture = Fixture::new().await;

    let err = fixture
        .ctx
        .shell("build", fixture.shell("true").output("target/report.html"))
        .await
        .expect_err("declared output never appeared");

    assert!(matches!(
        err,
        EngineError::MissingArtifact { ref step, ref pattern }
            if step == "build" && pattern == "target/report.html"
    ));

    let steps = fixture
        .store
        .list_steps(fixture.run_id)
        .await
        .expect("steps");
    assert_eq!(steps[0].status.state, StepStatus::Failed);
}

#[tokio::test]
async fn a_failed_step_still_yields_the_files_it_did_produce() {
    let mut fixture = Fixture::new().await;

    let err = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'partial' > build.log; exit 1")
                .output("build.log"),
        )
        .await
        .expect_err("command failed");

    // The command failure is what surfaces, not an artifact error.
    assert!(matches!(err, EngineError::Operation(_)));

    let artifacts = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list");
    assert_eq!(artifacts.len(), 1);
    assert_eq!(artifacts[0].name, "build.log");
    assert_eq!(artifacts[0].size_bytes, 7);
}

#[tokio::test]
async fn a_failed_step_tolerates_an_output_that_matched_nothing() {
    let mut fixture = Fixture::new().await;

    let err = fixture
        .ctx
        .shell("build", fixture.shell("exit 1").output("report.html"))
        .await
        .expect_err("command failed");

    assert!(matches!(err, EngineError::Operation(_)));
    assert!(
        fixture
            .store
            .list_artifacts_for_run(fixture.run_id)
            .await
            .expect("list")
            .is_empty()
    );
}

/// A consumer config holding a raw input, as a config stored by an older
/// version or built by hand would. Handles cannot express these cases.
fn with_raw_input(mut config: ShellConfig, step: &str, name: &str) -> ShellConfig {
    config.inputs.push(ArtifactInput::new(step, name));
    config
}

#[tokio::test]
async fn an_unresolvable_input_fails_the_step_before_the_command_runs() {
    let mut fixture = Fixture::new().await;

    let err = fixture
        .ctx
        .shell(
            "publish",
            with_raw_input(
                fixture.shell("printf 'ran' > marker.txt"),
                "build",
                "report.txt",
            ),
        )
        .await
        .expect_err("input does not exist");

    assert!(matches!(
        err,
        EngineError::ArtifactNotFound { ref step, ref name }
            if step == "build" && name == "report.txt"
    ));
    assert!(
        !fixture.work_dir.path().join("marker.txt").exists(),
        "the command must not have run"
    );
}

#[tokio::test]
async fn an_input_produced_by_a_later_step_is_not_visible() {
    let mut fixture = Fixture::new().await;

    // Position 0 declares an input produced at position 1: nothing can satisfy
    // it, because resolution only looks backwards.
    let err = fixture
        .ctx
        .shell(
            "publish",
            with_raw_input(fixture.shell("true"), "build", "report.txt"),
        )
        .await
        .expect_err("nothing produced yet");
    assert!(matches!(err, EngineError::ArtifactNotFound { .. }));

    fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'x' > report.txt")
                .output("report.txt"),
        )
        .await
        .expect("producer");
}

#[tokio::test]
async fn the_same_file_name_from_two_steps_is_kept_separately() {
    let mut fixture = Fixture::new().await;

    fixture
        .ctx
        .shell(
            "build",
            fixture.shell("printf 'one' > out.txt").output("out.txt"),
        )
        .await
        .expect("first");
    fixture
        .ctx
        .shell(
            "rebuild",
            fixture.shell("printf 'two' > out.txt").output("out.txt"),
        )
        .await
        .expect("second");

    let artifacts = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list");

    assert_eq!(artifacts.len(), 2);
    assert_ne!(artifacts[0].step_id, artifacts[1].step_id);
}

#[tokio::test]
async fn an_explicit_content_type_wins_over_the_guess() {
    let mut fixture = Fixture::new().await;

    fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf '{}' > data.txt")
                .output_typed("data.txt", "application/json"),
        )
        .await
        .expect("step");

    let artifacts = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list");

    assert_eq!(artifacts[0].content_type, "application/json");
}

#[tokio::test]
async fn a_file_name_outside_the_whitelist_is_refused() {
    let mut fixture = Fixture::new().await;

    let err = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'x' > 'rapport été.html'")
                .output("rapport été.html"),
        )
        .await
        .expect_err("name is not in the whitelist");

    assert!(matches!(err, EngineError::Artifact(_)));
}

#[tokio::test]
async fn a_step_without_artifacts_runs_without_a_backend() {
    let mut fixture = Fixture::without_artifact_storage().await;

    let output = fixture
        .ctx
        .shell("greet", fixture.shell("echo hello"))
        .await
        .expect("step");

    assert_eq!(output.stdout(), "hello");
}

#[tokio::test]
async fn declaring_an_output_without_a_backend_fails_explicitly() {
    let mut fixture = Fixture::without_artifact_storage().await;

    let err = fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("printf 'x' > report.txt")
                .output("report.txt"),
        )
        .await
        .expect_err("no artifact storage");

    assert!(matches!(err, EngineError::ArtifactsUnavailable(_)));
}

#[tokio::test]
async fn put_and_get_artifact_roundtrip_for_custom_operations() {
    let mut fixture = Fixture::new().await;

    // A shell step gives us a persisted step to hang the artifact on.
    let generate = fixture
        .ctx
        .shell("generate", fixture.shell("true"))
        .await
        .expect("step");

    let summary = fixture
        .ctx
        .put_artifact(&generate, "summary.json", None, br#"{"ok":true}"#.to_vec())
        .await
        .expect("put");
    assert_eq!(summary.step(), "generate");
    assert_eq!(summary.name(), "summary.json");

    let stored = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list");
    assert_eq!(stored.len(), 1);
    assert_eq!(stored[0].content_type, "application/json");
    let steps = fixture
        .store
        .list_steps(fixture.run_id)
        .await
        .expect("steps");
    assert_eq!(stored[0].step_id, steps[0].id);

    let bytes = fixture.ctx.get_artifact(&summary).await.expect("get");

    assert_eq!(bytes, br#"{"ok":true}"#);
}

#[tokio::test]
async fn get_artifact_on_a_declared_but_missing_file_reports_not_found() {
    let mut fixture = Fixture::new().await;

    // `*.json` is declared and matches `a.json`; `missing.json` fits the
    // declaration but was never produced.
    let build = fixture
        .ctx
        .shell(
            "build",
            fixture.shell("printf '{}' > a.json").output("*.json"),
        )
        .await
        .expect("step");
    let missing = build
        .artifact("missing.json")
        .expect("fits the declaration");

    let err = fixture
        .ctx
        .get_artifact(&missing)
        .await
        .expect_err("unknown artifact");

    assert!(matches!(
        err,
        EngineError::ArtifactNotFound { ref step, ref name }
            if step == "build" && name == "missing.json"
    ));
}

#[tokio::test]
async fn put_artifact_refuses_a_duplicate_name_on_the_same_step() {
    let mut fixture = Fixture::new().await;

    let generate = fixture
        .ctx
        .shell("generate", fixture.shell("true"))
        .await
        .expect("step");

    fixture
        .ctx
        .put_artifact(&generate, "a.json", None, b"{}".to_vec())
        .await
        .expect("first");

    let err = fixture
        .ctx
        .put_artifact(&generate, "a.json", None, b"{}".to_vec())
        .await
        .expect_err("duplicate");

    assert!(matches!(err, EngineError::Store(_)));
}

#[tokio::test]
async fn put_artifact_without_a_backend_fails_explicitly() {
    let mut fixture = Fixture::without_artifact_storage().await;

    let generate = fixture
        .ctx
        .shell("generate", fixture.shell("true"))
        .await
        .expect("step");

    let err = fixture
        .ctx
        .put_artifact(&generate, "a.json", None, b"{}".to_vec())
        .await
        .expect_err("no artifact storage");

    assert!(matches!(err, EngineError::ArtifactsUnavailable(_)));
}

#[tokio::test]
async fn an_empty_file_is_a_valid_artifact() {
    let mut fixture = Fixture::new().await;

    fixture
        .ctx
        .shell(
            "build",
            fixture.shell("touch empty.txt").output("empty.txt"),
        )
        .await
        .expect("step");

    let artifacts = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list");

    assert_eq!(artifacts[0].size_bytes, 0);
    assert_eq!(
        artifacts[0].sha256,
        "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    );
}

#[tokio::test]
async fn a_directory_matching_a_pattern_is_ignored() {
    let mut fixture = Fixture::new().await;

    // `out.d` is a directory and `out.f` a file; only the file is an artifact.
    fixture
        .ctx
        .shell(
            "build",
            fixture
                .shell("mkdir -p out.d && printf 'x' > out.f")
                .output("out.*"),
        )
        .await
        .expect("step");

    let artifacts = fixture
        .store
        .list_artifacts_for_run(fixture.run_id)
        .await
        .expect("list");

    assert_eq!(artifacts.len(), 1);
    assert_eq!(artifacts[0].name, "out.f");
}