tests/release_events.rs
//! Inject-and-assert integration test for the release-op DAG
//! (`release_events`, Epic B / B1+B2).
//!
//! Two independent checks against a **temp** warehouse (never the live
//! server's):
//!
//! 1. `injected_events_land_exactly` — drive the emitter + append + read
//! path with INJECTED components/ops/phases (no cargo, fully
//! deterministic) and assert the exact rows landed: run_id grouping,
//! `seq` monotonic, `depends_on` edges correct, the statuses we wrote,
//! `elapsed_ms` only on `end` rows. This is the Law #1 assertion.
//!
//! 2. `run_pipeline_records_release_events` — run the real
//! `run_pipeline` over two trivial crates in dry-run and assert it
//! emitted a structurally-correct event DAG for the run (run/start →
//! per-repo test/bench/gate → run/end), with `widget` carrying a
//! `depends_on = [acme]` edge taken from the dep graph.
use std::path::Path;
use nornir::release::pipeline::{self, PipelineOptions};
use nornir::warehouse::dep_graph;
use nornir::warehouse::iceberg::IcebergWarehouse;
use nornir::warehouse::release_events::{
EventSelector, ReleaseEventEmitter, ReleaseEventRow, append_release_events, phase,
query_release_events, status,
};
use nornir::workspace::WorkspaceDescriptor;
fn write(p: &Path, body: &str) {
if let Some(parent) = p.parent() {
std::fs::create_dir_all(parent).unwrap();
}
std::fs::write(p, body).unwrap();
}
fn git_init(root: &Path) {
nornir::gitio::init(root).unwrap();
nornir::gitio::commit_all(root, "init").unwrap();
}
#[test]
fn injected_events_land_exactly() {
let tmp = tempfile::tempdir().unwrap();
let wh_root = tmp.path().join("warehouse-root");
let wh = IcebergWarehouse::open(&wh_root).unwrap();
// ── INJECT a known run: znippy (root) → holger (depends on znippy) ──
// Drive the real emitter so `seq` is minted exactly as production does.
let emitter = ReleaseEventEmitter::new("run-INJ");
let znippy_deps: Vec<String> = vec![]; // root: no in-workspace producers
let holger_deps: Vec<String> = vec!["znippy".into()];
// run/start
let e_run_start = emitter.row(
"ws",
"",
"run",
phase::START,
status::RUNNING,
"order: znippy → holger",
None,
None,
);
// znippy test start/end (ok)
let e_zt_start = emitter.row(
"znippy",
"znippy",
"test",
phase::START,
status::RUNNING,
"",
Some(znippy_deps.clone()),
None,
);
let e_zt_end = emitter.row(
"znippy",
"znippy",
"test",
phase::END,
status::OK,
"12 passed, 0 failed",
Some(znippy_deps.clone()),
Some(420),
);
// znippy gate end (ok)
let e_zg_end = emitter.row(
"znippy",
"znippy",
"gate",
phase::END,
status::OK,
"succeeded_dry_run",
Some(znippy_deps.clone()),
None,
);
// holger test start/end (FAILS) — a partial release that stalls here.
let e_ht_start = emitter.row(
"holger",
"holger",
"test",
phase::START,
status::RUNNING,
"",
Some(holger_deps.clone()),
None,
);
let e_ht_end = emitter.row(
"holger",
"holger",
"test",
phase::END,
status::FAIL,
"8 passed, 1 failed",
Some(holger_deps.clone()),
Some(999),
);
// run/end (fail)
let e_run_end = emitter.row(
"ws",
"",
"run",
phase::END,
status::FAIL,
"stalled at holger",
None,
None,
);
// Append deliberately OUT of seq order to prove the reader re-sorts.
let injected = vec![
e_run_end.clone(),
e_ht_end.clone(),
e_zt_start.clone(),
e_run_start.clone(),
e_zg_end.clone(),
e_ht_start.clone(),
e_zt_end.clone(),
];
wh.block_on(append_release_events(&wh, &injected)).unwrap();
// ── reopen + read back by run_id (exercises the durable read path) ──
drop(wh);
let wh2 = IcebergWarehouse::open(&wh_root).unwrap();
let got = wh2
.block_on(query_release_events(
&wh2,
&EventSelector::Run("run-INJ".into()),
))
.unwrap();
// Exactly the 7 rows we injected, all under the one run_id.
assert_eq!(got.len(), 7, "expected exactly the injected rows: {got:#?}");
assert!(got.iter().all(|r| r.run_id == "run-INJ"), "run_id grouping");
// `seq` is monotonic 0..=6 in read order (the reader sorts by (run_id, seq)).
let seqs: Vec<i64> = got.iter().map(|r| r.seq).collect();
assert_eq!(
seqs,
vec![0, 1, 2, 3, 4, 5, 6],
"seq monotonic + reader-sorted"
);
// The injected emit order == seq order, so positionally:
let expected = [
&e_run_start,
&e_zt_start,
&e_zt_end,
&e_zg_end,
&e_ht_start,
&e_ht_end,
&e_run_end,
];
for (i, (g, e)) in got.iter().zip(expected).enumerate() {
assert_eq!(g, e, "row {i} must round-trip byte-for-byte");
}
// depends_on edges: znippy is a root (empty list, NOT null because we passed
// Some(vec![])); holger waited on znippy; the run/* rows carry null.
let znippy_test = got
.iter()
.find(|r| r.component == "znippy" && r.op == "test" && r.phase == phase::START)
.unwrap();
assert_eq!(
znippy_test.depends_on,
Some(vec![]),
"root component → empty dep list"
);
let holger_test = got
.iter()
.find(|r| r.component == "holger" && r.op == "test" && r.phase == phase::END)
.unwrap();
assert_eq!(
holger_test.depends_on,
Some(vec!["znippy".to_string()]),
"holger depends_on znippy"
);
let run_start = got
.iter()
.find(|r| r.op == "run" && r.phase == phase::START)
.unwrap();
assert_eq!(
run_start.depends_on, None,
"run envelope carries no dep edges"
);
// statuses: the failure is recorded exactly where we injected it.
assert_eq!(holger_test.status, status::FAIL);
let run_end = got
.iter()
.find(|r| r.op == "run" && r.phase == phase::END)
.unwrap();
assert_eq!(run_end.status, status::FAIL, "run rolled up the stall");
assert!(
got.iter()
.any(|r| r.component == "znippy" && r.op == "gate" && r.status == status::OK)
);
// elapsed_ms is present only on the `end` rows we stamped it on.
assert_eq!(holger_test.elapsed_ms, Some(999));
assert_eq!(
znippy_test.elapsed_ms, None,
"start rows carry no elapsed_ms"
);
let zt_end = got
.iter()
.find(|r| r.component == "znippy" && r.op == "test" && r.phase == phase::END)
.unwrap();
assert_eq!(zt_end.elapsed_ms, Some(420));
// Repo-scoped read crosses runs but filters by repo: only holger's 2 rows.
let holger_rows = wh2
.block_on(query_release_events(
&wh2,
&EventSelector::Repo("holger".into()),
))
.unwrap();
assert_eq!(
holger_rows.len(),
2,
"repo scope returns just holger's events"
);
assert!(holger_rows.iter().all(|r| r.repo == "holger"));
}
#[test]
fn injected_second_run_stays_isolated() {
// Two runs in the same warehouse must not bleed into each other.
let tmp = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(&tmp.path().join("wh")).unwrap();
let mk = |run: &str, comp: &str| ReleaseEventRow {
run_id: run.into(),
seq: 0,
ts_micros: 1,
component: comp.into(),
repo: comp.into(),
op: "test".into(),
phase: phase::START.into(),
status: status::RUNNING.into(),
detail: String::new(),
depends_on: None,
elapsed_ms: None,
};
wh.block_on(append_release_events(&wh, &[mk("runA", "x")]))
.unwrap();
wh.block_on(append_release_events(&wh, &[mk("runB", "y")]))
.unwrap();
let a = wh
.block_on(query_release_events(
&wh,
&EventSelector::Run("runA".into()),
))
.unwrap();
let b = wh
.block_on(query_release_events(
&wh,
&EventSelector::Run("runB".into()),
))
.unwrap();
assert_eq!(a.len(), 1);
assert_eq!(b.len(), 1);
assert_eq!(a[0].component, "x");
assert_eq!(b[0].component, "y");
let all = wh
.block_on(query_release_events(&wh, &EventSelector::All))
.unwrap();
assert_eq!(all.len(), 2, "All selector sees both runs");
}
#[test]
fn run_pipeline_records_release_events() {
let tmp = tempfile::tempdir().unwrap();
// repo A: produces crate `acme-lib`
let a = tmp.path().join("acme");
write(
&a.join("Cargo.toml"),
r#"[package]
name = "acme-lib"
version = "0.1.0"
edition = "2021"
publish = false
[lib]
path = "src/lib.rs"
"#,
);
write(&a.join("src/lib.rs"), "pub fn one() -> u32 { 1 }\n");
git_init(&a);
// repo B: consumes acme-lib via path dep
let b = tmp.path().join("widget");
write(
&b.join("Cargo.toml"),
&format!(
r#"[package]
name = "widget-lib"
version = "0.1.0"
edition = "2021"
publish = false
[lib]
path = "src/lib.rs"
[dependencies]
acme-lib = {{ path = "{}" }}
"#,
a.display()
),
);
write(
&b.join("src/lib.rs"),
"pub fn two() -> u32 { acme_lib::one() + 1 }\n",
);
git_init(&b);
let toml = format!(
r#"[workspace]
name = "acme-suite"
[repos.acme]
path = "{}"
[repos.widget]
path = "{}"
"#,
a.display(),
b.display()
);
let toml_path = tmp.path().join("nornir-workspace.toml");
std::fs::write(&toml_path, toml).unwrap();
let desc = WorkspaceDescriptor::load(&toml_path).unwrap();
let graph = dep_graph::build(&desc).unwrap();
let wh_root = tmp.path().join("warehouse-root");
let wh = IcebergWarehouse::open(&wh_root).unwrap();
let dep_snap = wh
.block_on(dep_graph::record_dep_graph(&wh, "acme-suite", &graph))
.unwrap();
let report = wh
.block_on(pipeline::run_pipeline(
&wh,
&desc,
&graph,
&PipelineOptions {
dry_run: true,
dep_graph_snapshot_id: dep_snap,
refuse_if_dirty: true,
progress: None,
gate_config: None,
},
))
.unwrap();
// ── reopen, read the events the run recorded ───────────────────────
drop(wh);
let wh2 = IcebergWarehouse::open(&wh_root).unwrap();
let events = wh2
.block_on(query_release_events(
&wh2,
&EventSelector::Run(report.release_id.to_string()),
))
.unwrap();
assert!(
!events.is_empty(),
"run_pipeline must record release_events"
);
// All under the run's release_id, and seq is strictly increasing.
assert!(
events
.iter()
.all(|r| r.run_id == report.release_id.to_string())
);
let seqs: Vec<i64> = events.iter().map(|r| r.seq).collect();
assert!(
seqs.windows(2).all(|w| w[0] < w[1]),
"seq strictly monotonic: {seqs:?}"
);
// The run envelope brackets everything.
assert_eq!(events.first().unwrap().op, "run");
assert_eq!(events.first().unwrap().phase, phase::START);
assert_eq!(events.last().unwrap().op, "run");
assert_eq!(events.last().unwrap().phase, phase::END);
assert_eq!(
events.last().unwrap().status,
status::OK,
"green dry-run → run/end ok"
);
// Each repo got a test/start, test/end and a gate/end. acme before widget
// (build order), proven by seq.
let acme_test_end = events
.iter()
.find(|r| r.component == "acme" && r.op == "test" && r.phase == phase::END)
.unwrap();
let widget_test_start = events
.iter()
.find(|r| r.component == "widget" && r.op == "test" && r.phase == phase::START)
.unwrap();
assert!(
acme_test_end.seq < widget_test_start.seq,
"deps-first: acme tested before widget"
);
assert_eq!(acme_test_end.status, status::OK);
// Per-repo events carry a real (possibly empty) `depends_on` list — never a
// null (the run envelope rows are the only nulls). The exact edge content is
// asserted byte-for-byte in `injected_events_land_exactly`, where the test
// owns the dep graph; here we assert the wiring took the graph's view of
// `widget`'s dependencies (which, for whatever the fixture's `cargo
// metadata` resolved, must match what the graph itself reports).
let acme_test_start = events
.iter()
.find(|r| r.component == "acme" && r.op == "test" && r.phase == phase::START)
.unwrap();
assert!(
widget_test_start.depends_on.is_some(),
"widget events carry a dep list, not null"
);
assert!(
acme_test_start.depends_on.is_some(),
"acme events carry a dep list, not null"
);
// What the graph reports for widget IS what the event recorded.
let mut graph_widget_deps: Vec<String> = graph
.dependencies_of("widget")
.iter()
.map(|e| e.to.clone())
.collect();
graph_widget_deps.sort();
graph_widget_deps.dedup();
assert_eq!(
widget_test_start.depends_on.clone().unwrap(),
graph_widget_deps,
"the emitted depends_on must equal the dep graph's dependencies_of(widget)"
);
// Both repos reached a passing gate/end.
for repo in ["acme", "widget"] {
let g = events
.iter()
.find(|r| r.component == repo && r.op == "gate" && r.phase == phase::END)
.unwrap();
assert_eq!(
g.status,
status::OK,
"{repo} gate must be ok in a green dry-run"
);
assert_eq!(g.detail, "succeeded_dry_run");
}
// Repo-scoped read returns only that repo's events (no run envelope rows).
let acme_only = wh2
.block_on(query_release_events(
&wh2,
&EventSelector::Repo("acme".into()),
))
.unwrap();
assert!(!acme_only.is_empty());
assert!(acme_only.iter().all(|r| r.repo == "acme"));
}