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

#![cfg(all(feature = "catalog", feature = "serve-history-sqlite"))]
//! End-to-end for `faucet history` (#391): seed run records into a real sqlite
//! catalog store, then drive `history::run` over the table, `--json`, and
//! `--row` paths. Exercises the command's connect → fetch → filter → render
//! wiring that the pure unit tests can't reach.

use chrono::Utc;
use faucet_cli::cli::HistoryArgs;
use faucet_cli::serve::history::{InvocationRecord, RunRecord, RunStatus};
use std::path::PathBuf;

fn args(config: PathBuf, row: Option<String>, json: bool) -> HistoryArgs {
    HistoryArgs {
        config: Some(config),
        env_file: None,
        no_env_file: true,
        profile: None,
        limit: 20,
        row,
        json,
    }
}

fn record(id: &str, row_id: &str) -> RunRecord {
    let t = Utc::now();
    RunRecord {
        run_id: id.into(),
        name: Some("demo".into()),
        labels: Default::default(),
        status: RunStatus::Completed,
        submitted_at: t,
        started_at: Some(t),
        finished_at: Some(t),
        elapsed_secs: Some(1.0),
        records_written: 10,
        invocations: vec![InvocationRecord {
            row_id: row_id.into(),
            parent_record_key: None,
            records_written: 10,
            error: None,
        }],
        error: None,
        idempotency_key: None,
        doctor_report: None,
        config_body: None,
        config_format: None,
        timeout_secs: None,
        clock: None,
        attempt: 0,
        replay_of: None,
        callback: None,
    }
}

#[tokio::test]
async fn history_reads_seeded_sqlite_catalog() {
    let dir = tempfile::tempdir().unwrap();
    let db = dir.path().join("cat.db");
    let cfg = dir.path().join("faucet.yaml");
    std::fs::write(
        &cfg,
        format!(
            "version: 1\nname: demo\ncatalog:\n  url: \"sqlite:{db}\"\npipeline:\n  source: {{ type: rest, config: {{ path: /x }} }}\n  sink: {{ type: jsonl, config: {{ path: o }} }}\n",
            db = db.display(),
        ),
    )
    .unwrap();

    // Seed two run records under different matrix rows.
    let handle = faucet_cli::catalog::connect_from_spec(&faucet_cli::catalog::CatalogSpec {
        url: format!("sqlite:{}", db.display()),
        sample_records: 10,
    })
    .await
    .unwrap();
    handle.store.upsert(&record("run-a", "us")).await.unwrap();
    handle.store.upsert(&record("run-b", "eu")).await.unwrap();

    // Table, JSON, and --row filter all drive `run()` to completion.
    faucet_cli::commands::history::run(args(cfg.clone(), None, false))
        .await
        .expect("table render");
    faucet_cli::commands::history::run(args(cfg.clone(), None, true))
        .await
        .expect("json render");
    faucet_cli::commands::history::run(args(cfg.clone(), Some("eu".into()), false))
        .await
        .expect("row filter");
}