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

#![cfg(feature = "lineage")]
//! End-to-end: a CSV→JSONL run with file lineage emits START + COMPLETE, and a
//! run with an unreachable lineage backend still succeeds (lineage never fails
//! a run).

use std::io::Write;
use std::path::PathBuf;

/// `RunArgs` does not derive `Default`, so construct it explicitly with the
/// given config path and the field defaults that match `cli/src/cli.rs`.
fn run_args(config: PathBuf) -> faucet_cli::cli::RunArgs {
    faucet_cli::cli::RunArgs {
        config: Some(config),
        no_env_file: true,
        ..Default::default()
    }
}

#[tokio::test]
async fn run_emits_start_and_complete_to_file() {
    let dir = tempfile::tempdir().unwrap();
    let input = dir.path().join("in.csv");
    let output = dir.path().join("out.jsonl");
    let ol = dir.path().join("lineage.jsonl");
    std::fs::write(&input, "id,name\n1,alice\n2,bob\n").unwrap();
    let cfg_path = dir.path().join("pipeline.yaml");
    let mut f = std::fs::File::create(&cfg_path).unwrap();
    write!(
        f,
        r#"version: 1
name: t
lineage:
  namespace: test
  include_schema_facet: true
  transport: {{ type: file, config: {{ path: {ol} }} }}
pipeline:
  source: {{ type: csv, config: {{ path: {input} }} }}
  sink:   {{ type: jsonl, config: {{ path: {output} }} }}
"#,
        ol = ol.display(),
        input = input.display(),
        output = output.display()
    )
    .unwrap();

    faucet_cli::commands::run::run(run_args(cfg_path))
        .await
        .unwrap();

    let body = std::fs::read_to_string(&ol).unwrap();
    let types: Vec<String> = body
        .lines()
        .map(|l| {
            serde_json::from_str::<serde_json::Value>(l).unwrap()["eventType"]
                .as_str()
                .unwrap()
                .to_string()
        })
        .collect();
    assert!(types.contains(&"START".to_string()));
    assert!(types.contains(&"COMPLETE".to_string()));
}

#[tokio::test]
async fn run_succeeds_even_when_lineage_backend_is_unreachable() {
    // Acceptance criterion: lineage emission NEVER fails a run. Point the HTTP
    // transport at a dead port; the CSV→JSONL run must still succeed.
    let dir = tempfile::tempdir().unwrap();
    let input = dir.path().join("in.csv");
    let output = dir.path().join("out.jsonl");
    std::fs::write(&input, "id\n1\n").unwrap();
    let cfg_path = dir.path().join("pipeline.yaml");
    let mut f = std::fs::File::create(&cfg_path).unwrap();
    write!(
        f,
        r#"version: 1
name: t
lineage:
  namespace: test
  transport: {{ type: http, config: {{ url: "http://127.0.0.1:1/api/v1/lineage", timeout_secs: 1 }} }}
pipeline:
  source: {{ type: csv, config: {{ path: {input} }} }}
  sink:   {{ type: jsonl, config: {{ path: {output} }} }}
"#,
        input = input.display(),
        output = output.display()
    )
    .unwrap();

    faucet_cli::commands::run::run(run_args(cfg_path))
        .await
        .expect("run must succeed despite an unreachable lineage backend");
    assert!(output.exists());
}