crates/knut-bifrost/tests/watch_cli_iceberg.rs
//! **`bifrost watch --warehouse …` — the CLI front door to the Iceberg → graph
//! streamer, driven as a real subprocess.**
//!
//! The library path (`IcebergSource` + `Listener` + the delta scan) already has
//! its live-FalkorDB proof in `delta_iceberg_falkordb.rs`. What THIS file proves
//! is the piece that was missing until the `watch` subcommand existed (§g5: "the
//! entire gap is one source constructor and a `--warehouse` flag"): that the
//! **binary** parses `watch`, builds an `IcebergSource` over a skade warehouse,
//! and runs the SAME `drive` loop the file-drop `run` path uses — including the
//! durable per-table snapshot cursor that makes a re-run a *delta*, not a
//! re-read.
//!
//! It runs the built `bifrost` binary (`CARGO_BIN_EXE_bifrost`) in `--dry-run`,
//! so it needs **no FalkorDB and no container** — the NullSink counts the
//! mutation instead of writing it, and the assertions read those counts back out
//! of the `--json-logs` stream. That keeps this in the DEFAULT (non-`--ignored`)
//! set: the CLI wiring is provable on any box.
//!
//! Run:
//! ```bash
//! cargo test -p knut-bifrost --features "cli skade" --test watch_cli_iceberg
//! ```
#![cfg(all(feature = "skade", feature = "cli"))]
use std::process::Command;
use std::sync::Arc;
use skade::arrow_array::{Int64Array, RecordBatch as SkadeBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
const MAPPING: &str = r#"
version: 1
tables:
- table: people
mode: incremental
pipeline:
- vertex:
label: Person
id_column: id
properties: [name]
"#;
fn people_batch(ids: &[i64]) -> SkadeBatch {
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
SkadeBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(ids.to_vec())),
Arc::new(StringArray::from(names)),
],
)
.unwrap()
}
/// Append `ids` as one new Iceberg snapshot via skade, dropping the warehouse
/// handle before returning so the `IcebergSource` (single-open redb) can reopen.
async fn append_iceberg(dir: &std::path::Path, ids: &[i64]) {
let wh = skade::Warehouse::open(dir).await.unwrap();
let batch = people_batch(ids);
let mut table = wh.table_or_create("people", &batch.schema()).await.unwrap();
table.append(&[batch]).await.unwrap();
}
/// Pull the `vertices` count out of the single `dry_run_mutation` json log line
/// the NullSink emits for a round. Format-tolerant: it finds the `vertices`
/// field and reads the integer that follows, whatever the json punctuation.
fn dry_run_vertices(stdout: &str) -> Option<u64> {
let line = stdout.lines().find(|l| l.contains("dry_run_mutation"))?;
let after = &line[line.find("vertices")? + "vertices".len()..];
let digits: String = after
.chars()
.skip_while(|c| !c.is_ascii_digit())
.take_while(|c| c.is_ascii_digit())
.collect();
digits.parse().ok()
}
/// Run `bifrost watch --once --dry-run` against `warehouse`/`mapping`, returning
/// the child's captured stdout after asserting a clean exit.
fn watch_once_dry_run(
warehouse: &std::path::Path,
mapping: &std::path::Path,
state: &std::path::Path,
) -> String {
let out = Command::new(env!("CARGO_BIN_EXE_bifrost"))
.args(["watch", "--once", "--dry-run", "--json-logs"])
.arg("--warehouse")
.arg(warehouse)
.arg("--mapping")
.arg(mapping)
.arg("--state")
.arg(state)
.output()
.expect("spawn bifrost");
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
assert!(
out.status.success(),
"bifrost watch --once --dry-run exited non-zero\nstdout:\n{stdout}\nstderr:\n{stderr}",
);
// The json logs are on stdout (knut_tracing `fmt().json()`), but be lenient.
if stdout.contains("dry_run_mutation") {
stdout
} else {
format!("{stdout}{stderr}")
}
}
/// **RED-when-broken: the `watch` subcommand ingests an Iceberg snapshot through
/// the CLI, and a re-run after a new snapshot ingests only the DELTA.**
///
/// Round 1 over a 3-row snapshot must count 3 vertices; after appending a 2-row
/// snapshot, round 2 (same `--state` cursor, `mode: incremental`) must count
/// exactly 2 — proving the CLI persisted and advanced the per-table snapshot
/// cursor rather than re-reading the whole table.
#[tokio::test(flavor = "multi_thread")]
async fn watch_cli_ingests_a_snapshot_then_only_the_delta() {
let tmp = tempfile::tempdir().unwrap();
let warehouse = tmp.path().join("lake");
std::fs::create_dir_all(&warehouse).unwrap();
let mapping = tmp.path().join("mapping.yml");
std::fs::write(&mapping, MAPPING).unwrap();
let state = tmp.path().join("bifrost-state.json");
// ── snapshot A: 3 rows → round 1 counts 3 ────────────────────────────────
append_iceberg(&warehouse, &[1, 2, 3]).await;
let out1 = watch_once_dry_run(&warehouse, &mapping, &state);
assert_eq!(
dry_run_vertices(&out1),
Some(3),
"round 1 over the 3-row snapshot counts 3 vertices\n{out1}",
);
assert!(
state.exists(),
"the snapshot cursor was persisted to --state"
);
// ── snapshot B: 2 more rows → round 2 counts only the 2 new ones ─────────
append_iceberg(&warehouse, &[4, 5]).await;
let out2 = watch_once_dry_run(&warehouse, &mapping, &state);
assert_eq!(
dry_run_vertices(&out2),
Some(2),
"round 2 is a DELTA: only the 2 rows the new snapshot added\n{out2}",
);
}
/// **RED-when-broken: an idle round (no new snapshot) ships nothing.**
/// After the first round consumes the only snapshot, a second round against the
/// unchanged warehouse must count 0 — the cursor makes a re-poll a no-op, not a
/// re-read.
#[tokio::test(flavor = "multi_thread")]
async fn watch_cli_idle_round_ships_nothing() {
let tmp = tempfile::tempdir().unwrap();
let warehouse = tmp.path().join("lake");
std::fs::create_dir_all(&warehouse).unwrap();
let mapping = tmp.path().join("mapping.yml");
std::fs::write(&mapping, MAPPING).unwrap();
let state = tmp.path().join("bifrost-state.json");
append_iceberg(&warehouse, &[1, 2, 3]).await;
let _ = watch_once_dry_run(&warehouse, &mapping, &state); // consume snapshot A
let idle = watch_once_dry_run(&warehouse, &mapping, &state);
// No new snapshot → no dry_run_mutation line at all, or an explicit 0.
let v = dry_run_vertices(&idle);
assert!(
v.is_none() || v == Some(0),
"an idle re-poll ships nothing (got {v:?})\n{idle}",
);
}
/// **RED-when-broken: without `--warehouse`, `watch` is a usage error.**
/// A guard that the required flag really is required (clap parse), so the
/// subcommand can never silently run against a default path.
#[test]
fn watch_requires_a_warehouse() {
let out = Command::new(env!("CARGO_BIN_EXE_bifrost"))
.args(["watch", "--mapping", "/nonexistent/m.yml"])
.output()
.expect("spawn bifrost");
assert!(!out.status.success(), "watch with no --warehouse must fail");
let err = String::from_utf8_lossy(&out.stderr);
assert!(
err.contains("warehouse"),
"the error names the missing flag: {err}"
);
}