faucet-cli 1.8.1

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

#![cfg(all(feature = "catalog", feature = "serve-history-sqlite"))]
//! End-to-end for the config-change preview (#374): a successful `faucet run`
//! with a `catalog:` block records a config snapshot, and `faucet plan --diff`
//! then diffs the current config against it. Exercises the real command wiring
//! (`run`'s record hook + `plan --diff`) over a sqlite catalog on local disk.

use std::path::PathBuf;

fn run_args(config: PathBuf) -> faucet_cli::cli::RunArgs {
    faucet_cli::cli::RunArgs {
        config: Some(config),
        no_env_file: true,
        ..Default::default()
    }
}

fn plan_diff_args(config: PathBuf) -> faucet_cli::cli::PlanArgs {
    faucet_cli::cli::PlanArgs {
        config: Some(config),
        row: None,
        sample: None,
        live: false,
        limit: 10,
        json: false,
        diff: true,
        resolve_secrets: false,
        profile: None,
    }
}

#[tokio::test]
async fn run_records_snapshot_then_plan_diff_reads_it() {
    let dir = tempfile::tempdir().unwrap();
    let input = dir.path().join("in.csv");
    let output = dir.path().join("out.jsonl");
    let db = dir.path().join("cat.db");
    std::fs::write(&input, "id,name\n1,alice\n2,bob\n").unwrap();

    let cfg_path = dir.path().join("pipeline.yaml");
    std::fs::write(
        &cfg_path,
        format!(
            "version: 1\nname: e2ediff\ncatalog:\n  url: \"sqlite:{db}\"\npipeline:\n  source: {{ type: csv, config: {{ path: {input} }} }}\n  sink: {{ type: jsonl, config: {{ path: {output} }} }}\n",
            db = db.display(),
            input = input.display(),
            output = output.display(),
        ),
    )
    .unwrap();

    // A successful run must record a config snapshot (best-effort hook).
    faucet_cli::commands::run::run(run_args(cfg_path.clone()))
        .await
        .expect("run");

    let handle = faucet_cli::catalog::connect_from_spec(&faucet_cli::catalog::CatalogSpec {
        url: format!("sqlite:{}", db.display()),
        sample_records: 10,
    })
    .await
    .unwrap();
    let snap = handle
        .store
        .catalog_last_config_snapshot("e2ediff")
        .await
        .unwrap();
    assert!(snap.is_some(), "run should have recorded a config snapshot");
    let snap = snap.unwrap();
    assert_eq!(snap.pipeline, "e2ediff");
    assert!(!snap.rows.is_empty());

    // `plan --diff` now compares the current config against the recorded run.
    faucet_cli::commands::plan::run(plan_diff_args(cfg_path))
        .await
        .expect("plan --diff against recorded snapshot");
}