#![cfg(all(
feature = "catalog",
feature = "serve-history-sqlite",
feature = "source-csv",
feature = "sink-jsonl"
))]
use faucet_cli::cli::{
CatalogArgs, CatalogCommand, CatalogConfigArgs, CatalogDatasetsArgs, CatalogLineageArgs,
CatalogShowArgs,
};
use faucet_cli::commands::catalog as catalog_cmd;
use faucet_cli::serve::history::catalog::CatalogListFilter;
use std::path::{Path, PathBuf};
fn config_yaml(dir: &Path) -> String {
format!(
"version: 1\nname: catalog-cmd\ncatalog:\n url: \"sqlite:{dir}/cat.db\"\n sample_records: 10\npipeline:\n source: {{ type: csv, config: {{ path: \"{dir}/in.csv\" }} }}\n sink: {{ type: jsonl, config: {{ path: \"{dir}/out.jsonl\" }} }}\n",
dir = dir.display()
)
}
fn common(config: Option<PathBuf>, json: bool) -> CatalogConfigArgs {
CatalogConfigArgs {
config,
env_file: None,
no_env_file: true,
profile: None,
json,
}
}
async fn seed(dir: &Path) -> PathBuf {
let yaml = config_yaml(dir);
let config_path = dir.join("faucet.yaml");
std::fs::write(&config_path, &yaml).unwrap();
std::fs::write(dir.join("in.csv"), "id,name\n1,alice\n2,bob\n").unwrap();
faucet_cli::run_from_yaml_str(&yaml).await.expect("run 1");
std::fs::write(
dir.join("in.csv"),
"id,name,email\n1,alice,a@x.io\n2,bob,b@x.io\n",
)
.unwrap();
faucet_cli::run_from_yaml_str(&yaml).await.expect("run 2");
config_path
}
async fn source_dataset_id(dir: &Path) -> String {
let handle = faucet_cli::catalog::connect_from_spec(&faucet_cli::catalog::CatalogSpec {
url: format!("sqlite:{}/cat.db", dir.display()),
sample_records: 10,
})
.await
.unwrap();
let page = handle
.store
.catalog_list_datasets(&CatalogListFilter {
limit: 10,
..Default::default()
})
.await
.unwrap();
assert_eq!(page.datasets.len(), 2, "source + sink datasets seeded");
page.datasets
.iter()
.find(|d| d.kind == "csv")
.expect("csv dataset")
.id
.clone()
}
#[tokio::test(flavor = "multi_thread")]
async fn catalog_command_datasets_show_lineage_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let config = seed(dir.path()).await;
let id = source_dataset_id(dir.path()).await;
for (json, kind, q) in [
(false, None, None),
(true, None, None),
(false, Some("csv".to_string()), None),
(false, None, Some("out.jsonl".to_string())),
] {
catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Datasets(CatalogDatasetsArgs {
common: common(Some(config.clone()), json),
kind,
q,
limit: 50,
}),
})
.await
.expect("datasets");
}
for (shown, json) in [(id.clone(), false), (id[..8].to_string(), true)] {
catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Show(CatalogShowArgs {
id: shown,
common: common(Some(config.clone()), json),
}),
})
.await
.expect("show");
}
let err = catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Show(CatalogShowArgs {
id: "ffffffffffffffff".into(),
common: common(Some(config.clone()), false),
}),
})
.await
.unwrap_err();
assert!(err.to_string().contains("no catalogued dataset"), "{err}");
let err = catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Show(CatalogShowArgs {
id: "".into(),
common: common(Some(config.clone()), false),
}),
})
.await
.unwrap_err();
assert!(err.to_string().contains("ambiguous"), "{err}");
for (json, root) in [(false, None), (true, None), (false, Some(id.clone()))] {
catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Lineage(CatalogLineageArgs {
common: common(Some(config.clone()), json),
root,
depth: 3,
}),
})
.await
.expect("lineage");
}
catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Lineage(CatalogLineageArgs {
common: common(Some(config.clone()), false),
root: Some("ffffffffffffffff".into()),
depth: 3,
}),
})
.await
.expect("empty rooted lineage");
}
#[tokio::test(flavor = "multi_thread")]
async fn catalog_command_errors_without_a_catalog_block() {
let dir = tempfile::tempdir().unwrap();
let config = dir.path().join("plain.yaml");
std::fs::write(
&config,
"version: 1\npipeline:\n source: { type: csv, config: { path: ./in.csv } }\n sink: { type: jsonl, config: { path: ./out.jsonl } }\n",
)
.unwrap();
let err = catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Datasets(CatalogDatasetsArgs {
common: common(Some(config), false),
kind: None,
q: None,
limit: 10,
}),
})
.await
.unwrap_err();
assert!(err.to_string().contains("no `catalog:` block"), "{err}");
}
#[tokio::test(flavor = "multi_thread")]
async fn catalog_command_on_an_empty_store_prints_the_empty_message() {
let dir = tempfile::tempdir().unwrap();
let config = dir.path().join("faucet.yaml");
std::fs::write(
&config,
"version: 1\ncatalog: { url: memory }\npipeline:\n source: { type: csv, config: { path: ./in.csv } }\n sink: { type: jsonl, config: { path: ./out.jsonl } }\n",
)
.unwrap();
catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Datasets(CatalogDatasetsArgs {
common: common(Some(config.clone()), false),
kind: None,
q: None,
limit: 10,
}),
})
.await
.expect("empty datasets");
catalog_cmd::run(CatalogArgs {
command: CatalogCommand::Lineage(CatalogLineageArgs {
common: common(Some(config), false),
root: None,
depth: 3,
}),
})
.await
.expect("empty lineage");
}