faucet-cli 1.12.0

Config-driven CLI runner for faucet-stream pipelines (YAML / JSON, Meltano-style)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//! End-to-end: a real `faucet run` pipeline with a WASM transform, driving the
//! precompiled reference module at `examples/wasm-transforms/add_field.wasm`
//! (built from `examples/wasm-transforms/rust/`). Gated on `transform-wasm`.
#![cfg(feature = "transform-wasm")]

use assert_cmd::Command;
use std::fs;
use tempfile::TempDir;

/// The committed reference module (Rust → wasm): uppercases `name` and stamps
/// `wasm_processed: true`.
const ADD_FIELD_WASM: &[u8] = include_bytes!("../../examples/wasm-transforms/add_field.wasm");

#[test]
fn run_pipeline_with_wasm_transform() {
    let dir = TempDir::new().unwrap();
    let module = dir.path().join("add_field.wasm");
    fs::write(&module, ADD_FIELD_WASM).unwrap();

    let csv = dir.path().join("in.csv");
    fs::write(&csv, "name,id\nalice,1\nbob,2\n").unwrap();
    let out = dir.path().join("out.jsonl");

    let cfg = dir.path().join("pipeline.yaml");
    fs::write(
        &cfg,
        format!(
            r#"version: 1
name: wasm_e2e
pipeline:
  source: {{ type: csv, config: {{ path: {csv} }} }}
  transforms:
    - type: wasm
      config:
        module: {module}
  sink: {{ type: jsonl, config: {{ path: {out} }} }}
"#,
            csv = csv.display(),
            module = module.display(),
            out = out.display(),
        ),
    )
    .unwrap();

    Command::cargo_bin("faucet")
        .unwrap()
        .args(["run"])
        .arg(&cfg)
        .assert()
        .success();

    let written = fs::read_to_string(&out).unwrap();
    let lines: Vec<serde_json::Value> = written
        .lines()
        .filter(|l| !l.trim().is_empty())
        .map(|l| serde_json::from_str(l).unwrap())
        .collect();
    assert_eq!(lines.len(), 2, "two records expected, got: {written}");
    // The module uppercases `name` and adds `wasm_processed: true`.
    assert_eq!(lines[0]["name"], "ALICE");
    assert_eq!(lines[0]["wasm_processed"], true);
    assert_eq!(lines[1]["name"], "BOB");
    assert_eq!(lines[1]["wasm_processed"], true);
}

#[test]
fn run_fails_when_module_missing() {
    let dir = TempDir::new().unwrap();
    let csv = dir.path().join("in.csv");
    fs::write(&csv, "name,id\nalice,1\n").unwrap();
    let out = dir.path().join("out.jsonl");
    let cfg = dir.path().join("pipeline.yaml");
    fs::write(
        &cfg,
        format!(
            r#"version: 1
name: wasm_missing
pipeline:
  source: {{ type: csv, config: {{ path: {csv} }} }}
  transforms:
    - type: wasm
      config:
        module: {dir}/does-not-exist.wasm
  sink: {{ type: jsonl, config: {{ path: {out} }} }}
"#,
            csv = csv.display(),
            dir = dir.path().display(),
            out = out.display(),
        ),
    )
    .unwrap();

    Command::cargo_bin("faucet")
        .unwrap()
        .args(["run"])
        .arg(&cfg)
        .assert()
        .failure();
}