crates/knut-bifrost/tests/lifecycle.rs
//! Lifecycle test matrix: **create file → detect → ingest → teardown**, with
//! no container and no human. Each case stages parquet drops in a fresh temp
//! directory (setup), runs the bifröst `Listener` with a recording sink, and
//! asserts what was detected and ingested; the tempdir drop is teardown.
//!
//! The final test asserts on the **structured tracing events** the library
//! emits (`file_detected`, `table_synced`, `round_complete`) — proving the log
//! stream is a stable contract a test or bencher can drive headlessly.
use std::sync::{Arc, Mutex};
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use knut_bifrost::{
GraphMutation, Listener, MappingSpec, SinkReport, SyncMode, SyncState, TransformRegistry,
sink::GraphSink, source::FileSource,
};
/// Emit a functional-status row into the nornir test matrix. With the
/// `testmatrix` feature ON it records `ok`/`detail` for `check`; with the
/// feature OFF it is a no-op (and `nornir-testmatrix` is not built into this
/// test target). Component is the integration suite that ran.
#[cfg(feature = "testmatrix")]
fn fs(check: &str, ok: bool, detail: &str) {
nornir_testmatrix::functional_status("knut-bifrost-lifecycle", check, ok, detail);
}
#[cfg(not(feature = "testmatrix"))]
fn fs(_check: &str, _ok: bool, _detail: &str) {}
const MAPPING: &str = r#"
version: 1
tables:
- table: people
mode: MODE
pipeline:
- vertex:
label: Person
id_column: id
properties: [name]
"#;
fn people_batch(ids: &[i64]) -> RecordBatch {
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
RecordBatch::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()
}
/// Stage one parquet chunk under `<root>/people/<name>`.
fn drop_chunk(root: &std::path::Path, name: &str, ids: &[i64]) {
let dir = root.join("people");
std::fs::create_dir_all(&dir).unwrap();
graphar::io::write_chunk(
dir.join(name),
std::slice::from_ref(&people_batch(ids)),
&graphar::FileType::Parquet,
)
.unwrap();
}
/// A sink that records every mutation it receives — the "ingested" side.
#[derive(Clone, Default)]
struct RecordingSink {
vertices: Arc<Mutex<usize>>,
edges: Arc<Mutex<usize>>,
}
#[async_trait]
impl GraphSink for RecordingSink {
async fn apply(&mut self, mutation: &GraphMutation) -> knut_bifrost::Result<SinkReport> {
let v: usize = mutation.vertices.iter().map(|s| s.batch.num_rows()).sum();
let e: usize = mutation.edges.iter().map(|s| s.batch.num_rows()).sum();
*self.vertices.lock().unwrap() += v;
*self.edges.lock().unwrap() += e;
let d: usize = mutation.deletes.iter().map(|s| s.batch.num_rows()).sum();
Ok(SinkReport {
vertices_written: v,
edges_written: e,
deleted: d,
})
}
}
fn mapping_for(mode: SyncMode) -> MappingSpec {
let m = match mode {
SyncMode::Full => "full",
SyncMode::Incremental => "incremental",
SyncMode::Cdc => "cdc",
};
MappingSpec::from_yaml(&MAPPING.replace("MODE", m)).unwrap()
}
/// Build a listener over `root` with a fresh recording sink + state file.
fn listener(
root: &std::path::Path,
mode: SyncMode,
) -> (Listener<FileSource, RecordingSink>, RecordingSink) {
let sink = RecordingSink::default();
let listener = Listener::new(
mapping_for(mode),
TransformRegistry::new(),
FileSource::new(root),
sink.clone(),
SyncState::open(root.join(".state.json")).unwrap(),
);
(listener, sink)
}
// ── the matrix ──────────────────────────────────────────────────────────────
#[tokio::test]
async fn incremental_single_drop_then_idle() {
let dir = tempfile::tempdir().unwrap();
drop_chunk(dir.path(), "chunk0.parquet", &[1, 2, 3]);
let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);
let round = l.sync_once().await.unwrap();
assert_eq!(round.synced.len(), 1, "the drop was detected + ingested");
assert_eq!(round.synced[0].1, 3, "3 rows");
assert_eq!(*sink.vertices.lock().unwrap(), 3, "3 vertices ingested");
// Idle: no new file → nothing re-ingested.
let idle = l.sync_once().await.unwrap();
assert!(idle.synced.is_empty());
let v = *sink.vertices.lock().unwrap();
assert_eq!(v, 3, "no double-ingest");
fs(
"incremental_single_drop_then_idle",
round.synced.len() == 1 && round.synced[0].1 == 3 && v == 3 && idle.synced.is_empty(),
&format!("1 drop → 3 rows / {v} vertices, then idle (no double-ingest)"),
);
}
#[tokio::test]
async fn incremental_second_drop_is_delta_only() {
let dir = tempfile::tempdir().unwrap();
drop_chunk(dir.path(), "chunk0.parquet", &[1, 2]);
let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);
l.sync_once().await.unwrap();
assert_eq!(*sink.vertices.lock().unwrap(), 2);
// A second file appears → only its rows flow.
drop_chunk(dir.path(), "chunk1.parquet", &[3, 4, 5]);
let round = l.sync_once().await.unwrap();
assert_eq!(round.synced[0].1, 3, "only the new file's 3 rows");
let v = *sink.vertices.lock().unwrap();
assert_eq!(v, 5, "2 + 3 total");
fs(
"incremental_second_drop_is_delta_only",
round.synced[0].1 == 3 && v == 5,
&format!("second drop shipped only its 3 rows; {v} vertices total"),
);
}
#[tokio::test]
async fn full_mode_remerges_every_round() {
let dir = tempfile::tempdir().unwrap();
drop_chunk(dir.path(), "chunk0.parquet", &[1, 2]);
drop_chunk(dir.path(), "chunk1.parquet", &[3, 4]);
let (mut l, sink) = listener(dir.path(), SyncMode::Full);
let round = l.sync_once().await.unwrap();
assert_eq!(round.synced[0].1, 4, "both files read");
// Full re-reads every round (idempotent MERGE downstream).
let round2 = l.sync_once().await.unwrap();
assert_eq!(round2.synced[0].1, 4, "full re-reads");
let v = *sink.vertices.lock().unwrap();
assert_eq!(v, 8, "4 + 4 (MERGE dedupes in a real sink)");
fs(
"full_mode_remerges_every_round",
round.synced[0].1 == 4 && round2.synced[0].1 == 4 && v == 8,
&format!("full mode re-read 4 rows each round → {v} vertices"),
);
}
#[tokio::test]
async fn empty_directory_is_idle() {
let dir = tempfile::tempdir().unwrap();
let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);
let round = l.sync_once().await.unwrap();
assert!(round.synced.is_empty() && round.failed.is_empty());
let v = *sink.vertices.lock().unwrap();
assert_eq!(v, 0);
fs(
"empty_directory_is_idle",
round.synced.is_empty() && round.failed.is_empty() && v == 0,
"empty dir → idle round, 0 vertices",
);
}
#[tokio::test]
async fn multi_file_first_round_reads_all() {
let dir = tempfile::tempdir().unwrap();
for (i, name) in ["a.parquet", "b.parquet", "c.parquet"].iter().enumerate() {
drop_chunk(dir.path(), name, &[(i * 10) as i64, (i * 10 + 1) as i64]);
}
let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);
let round = l.sync_once().await.unwrap();
assert_eq!(round.synced[0].1, 6, "3 files × 2 rows");
let v = *sink.vertices.lock().unwrap();
assert_eq!(v, 6);
fs(
"multi_file_first_round_reads_all",
round.synced[0].1 == 6 && v == 6,
&format!("first round read all 3 files → {v} vertices"),
);
}
// ── AIP-82: the listener auto-emits asset events after a round ──────────────
#[tokio::test]
async fn listener_emits_asset_events_after_round() {
use knut_bifrost::{AssetEmitter, AssetEvent};
#[derive(Clone, Default)]
struct RecEmitter {
seen: Arc<Mutex<Vec<AssetEvent>>>,
}
#[async_trait]
impl AssetEmitter for RecEmitter {
async fn emit(&self, events: &[AssetEvent]) -> knut_bifrost::Result<()> {
self.seen.lock().unwrap().extend_from_slice(events);
Ok(())
}
}
let dir = tempfile::tempdir().unwrap();
drop_chunk(dir.path(), "chunk0.parquet", &[1, 2, 3]);
let emitter = RecEmitter::default();
let (l, _sink) = listener(dir.path(), SyncMode::Incremental);
let mut l = l.with_asset_emitter(Arc::new(emitter.clone()), "graph://demo/");
// First round ingests + emits one asset event for the `people` table.
l.sync_once().await.unwrap();
{
let seen = emitter.seen.lock().unwrap();
assert_eq!(seen.len(), 1, "one asset event per ingested table");
assert_eq!(seen[0].uri, "graph://demo/people");
assert_eq!(seen[0].extra["rows"], 3);
}
// Idle round: nothing ingested → nothing emitted.
l.sync_once().await.unwrap();
let n = emitter.seen.lock().unwrap().len();
assert_eq!(n, 1, "idle round emits no asset events");
fs(
"listener_emits_asset_events_after_round",
n == 1,
&format!("{n} AIP-82 asset event after ingest; idle round emits none"),
);
}
// ── the log contract (drive the real CLI, assert its JSON) ──────────────────
/// Drive the `bifrost` binary headlessly — create a file drop, run one
/// `--dry-run` round with `--json-logs`, and assert the structured lifecycle
/// events appear in order on stdout. This is the contract a test or bencher
/// keys off: no database, no human, no in-process subscriber races.
///
/// Requires `--features cli` (so the `bifrost` bin is built); gated to avoid a
/// missing-binary error on the default test run.
#[cfg(feature = "cli")]
#[test]
fn bifrost_cli_emits_lifecycle_json() {
use std::process::Command;
let dir = tempfile::tempdir().unwrap();
drop_chunk(dir.path(), "chunk0.parquet", &[1, 2, 3]);
let mapping = dir.path().join("mapping.yml");
std::fs::write(&mapping, MAPPING.replace("MODE", "incremental")).unwrap();
let out = Command::new(env!("CARGO_BIN_EXE_bifrost"))
.args(["--json-logs", "run"])
.arg("--mapping")
.arg(&mapping)
.arg("--files")
.arg(dir.path())
.args(["--dry-run", "--once"])
.output()
.expect("run bifrost");
assert!(
out.status.success(),
"bifrost exited non-zero: {:?}",
out.status
);
let logs = String::from_utf8(out.stdout).unwrap();
let events: Vec<String> = logs
.lines()
.filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
.filter_map(|v| v["fields"]["event"].as_str().map(String::from))
.collect();
for expected in [
"file_detected",
"files_read",
"dry_run_mutation",
"table_synced",
"round_complete",
] {
assert!(
events.contains(&expected.to_string()),
"missing event {expected}; got {events:?}"
);
}
let pos = |e: &str| events.iter().position(|x| x == e).unwrap();
let ordered =
pos("file_detected") < pos("table_synced") && pos("table_synced") < pos("round_complete");
assert!(
pos("file_detected") < pos("table_synced"),
"detect precedes ingest"
);
assert!(
pos("table_synced") < pos("round_complete"),
"ingest precedes round end"
);
fs(
"bifrost_cli_emits_lifecycle_json",
out.status.success() && ordered,
"CLI --json-logs emits file_detected→table_synced→round_complete in order",
);
}