facett-pipeline/tests/knut_facett_matrix.rs
//! **knut → facett connector test matrix.**
//!
//! This is the functional, multi-dimensional test of the *handoff* between
//! knut's pipeline model (`knut_pipelines::{DataflowGraph, ViewerState,
//! NodeStatus}`) and facett's visual (`facett_pipeline::PipelineView`). It is
//! the LAW-1 "real-in / real-out" proof for the connector: every cell builds a
//! REAL knut `DataflowGraph` + injects REAL run statuses/events/progress, maps it
//! through the SAME adapter korp uses, feeds it to `PipelineView`, and asserts
//! the rendered structure via the component's `state_json` (node count, each
//! node's status, the failed node's error text, the progress numbers) — never a
//! bare "didn't panic".
//!
//! ## Matrix dimensions (3 × 4 × 3 = 36 cells)
//! * **shape** — Linear · Diamond · WideFanOut
//! * **status** — AllPending · MidRunMixed · AllOk · OneFailed
//! * **progress** — None · Streaming · Complete
//!
//! Each cell asserts, against the rendered `state_json`:
//! 1. the node + edge counts match the graph,
//! 2. the per-node status counts match the injected statuses,
//! 3. when a node failed, its **error message** is surfaced in `errors[]`,
//! 4. the progress block matches the injected progress (or is null).
use facett_core::Facet;
use facett_pipeline::PipelineView;
use knut_pipelines::{
DataflowGraph, Dataset, EventLevel, ExecutionProgress, Flow, OutputType, PipelineRunEvent,
StageProgress, ViewerState, ViewerUpdate,
};
// ── matrix axes ──────────────────────────────────────────────────────────────
#[derive(Clone, Copy, Debug)]
enum Shape {
Linear,
Diamond,
WideFanOut,
}
#[derive(Clone, Copy, Debug)]
enum StatusMix {
AllPending,
MidRunMixed,
AllOk,
OneFailed,
}
#[derive(Clone, Copy, Debug)]
enum ProgressState {
NoProgress,
Streaming,
Complete,
}
impl Shape {
/// A real knut `DataflowGraph` for this topology. Names are deterministic so
/// assertions can address nodes by name.
fn graph(self) -> DataflowGraph {
let ds = |n: &str| Dataset::new(n, OutputType::MaterializedView);
match self {
// a → b → c → d
Shape::Linear => DataflowGraph::new()
.with_dataset(ds("a"))
.with_dataset(ds("b"))
.with_dataset(ds("c"))
.with_dataset(ds("d"))
.with_flow(Flow::new("f_a", "a", Vec::<String>::new()))
.with_flow(Flow::new("f_b", "b", ["a"]))
.with_flow(Flow::new("f_c", "c", ["b"]))
.with_flow(Flow::new("f_d", "d", ["c"])),
// src → {l, r} → sink (a diamond)
Shape::Diamond => DataflowGraph::new()
.with_dataset(ds("src"))
.with_dataset(ds("l"))
.with_dataset(ds("r"))
.with_dataset(ds("sink"))
.with_flow(Flow::new("f_src", "src", Vec::<String>::new()))
.with_flow(Flow::new("f_l", "l", ["src"]))
.with_flow(Flow::new("f_r", "r", ["src"]))
.with_flow(Flow::new("f_sink", "sink", ["l", "r"])),
// root → {c0, c1, c2, c3, c4} (wide fan-out)
Shape::WideFanOut => {
let mut g = DataflowGraph::new()
.with_dataset(ds("root"))
.with_flow(Flow::new("f_root", "root", Vec::<String>::new()));
for k in 0..5 {
let name = format!("c{k}");
g = g
.with_dataset(ds(&name))
.with_flow(Flow::new(format!("f_c{k}"), name, ["root"]));
}
g
}
}
}
fn node_count(self) -> usize {
match self {
Shape::Linear => 4,
Shape::Diamond => 4,
Shape::WideFanOut => 6,
}
}
/// Distinct (read → target) edges the adapter will draw.
fn edge_count(self) -> usize {
match self {
Shape::Linear => 3, // a→b, b→c, c→d
Shape::Diamond => 4, // src→l, src→r, l→sink, r→sink
Shape::WideFanOut => 5, // root→c0..c4
}
}
}
impl StatusMix {
/// Inject this status mix into a fresh `ViewerState` by streaming REAL
/// `ViewerUpdate::Event`s (running / completion / error) — exactly how a live
/// run drives the model — so the resulting `node_status` is what the viewer
/// would actually compute, not a hand-set map. Returns the name of the failed
/// node (if any) plus its error text, for the error-surfacing assertion.
fn apply(self, vs: &mut ViewerState, names: &[String]) -> Option<(String, String)> {
match self {
StatusMix::AllPending => None, // no events: everything stays Idle
StatusMix::AllOk => {
for n in names {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("Flow into {n} has COMPLETED"),
element: Some(n.clone()),
}));
}
None
}
StatusMix::MidRunMixed => {
// first done, second running, rest idle.
if let Some(n) = names.first() {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("{n} materialized"),
element: Some(n.clone()),
}));
}
if let Some(n) = names.get(1) {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("running flow into {n}"),
element: Some(n.clone()),
}));
}
None
}
StatusMix::OneFailed => {
// mark all-but-last done, then fail the last node with a real error.
let n = names.len();
for name in names.iter().take(n.saturating_sub(1)) {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("{name} materialized"),
element: Some(name.clone()),
}));
}
let failed = names.last().cloned().unwrap();
let err = format!("AnalysisException: column missing in {failed}");
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Error,
message: err.clone(),
element: Some(failed.clone()),
}));
Some((failed, err))
}
}
}
}
impl ProgressState {
fn apply(self, vs: &mut ViewerState, names: &[String]) {
match self {
ProgressState::NoProgress => {}
ProgressState::Streaming => {
vs.apply(ViewerUpdate::Progress(ExecutionProgress {
stages: vec![
StageProgress {
stage_id: 0,
num_tasks: 8,
num_completed_tasks: 8,
input_bytes_read: 2 * 1024 * 1024,
done: true,
},
StageProgress {
stage_id: 1,
num_tasks: 8,
num_completed_tasks: 2,
input_bytes_read: 1024 * 1024,
done: false,
},
],
num_inflight_tasks: 6,
}));
// A live streaming run has ALSO written rows and spans real
// wall-clock time — so its `ViewerState` carries `rows_written`
// (parsed from "wrote N rows" events) and timestamped log events,
// from which the adapter derives `rows` + `elapsed_s` and thus a
// throughput (`rows_per_s > 0`). Progress task/byte counts alone
// never yield those: `ExecutionProgress` carries neither rows nor
// timestamps. We seed both directly (public run-state fields)
// rather than through `ViewerState::apply`, because routing an
// event through `apply` attributes it to a node — via a raw
// substring scan that a single-letter dataset name (`a`/`l`/`r`)
// matches — and would flip that node's status, colliding with the
// independently-controlled status axis of the matrix. Direct
// seeding keeps rows+elapsed orthogonal to status.
if let Some(first) = names.first() {
// The completed first stage reported its real row count.
vs.rows_written.insert(first.clone(), 5_000);
}
// Two timestamped log events → a measured 4s wall-clock span
// (element-free + Info, so they touch neither status nor errors).
vs.events.push(PipelineRunEvent {
timestamp: Some("2026-01-01T00:00:00Z".into()),
level: EventLevel::Info,
message: "streaming run started".into(),
element: None,
});
vs.events.push(PipelineRunEvent {
timestamp: Some("2026-01-01T00:00:04Z".into()),
level: EventLevel::Info,
message: "streaming micro-batch flushed".into(),
element: None,
});
}
ProgressState::Complete => {
vs.apply(ViewerUpdate::Progress(ExecutionProgress {
stages: vec![StageProgress {
stage_id: 0,
num_tasks: 10,
num_completed_tasks: 10,
input_bytes_read: 4 * 1024 * 1024,
done: true,
}],
num_inflight_tasks: 0,
}));
}
}
}
}
// ── the connector: knut's PipelineAdapter → the normalized model → the view ──
//
// The handoff is now **engine-agnostic**: knut's `ViewerState` is mapped to
// facett's neutral `PipelineModel` by knut's `PipelineAdapter` impl
// (`to_pipeline_model`, feature `facett-adapter`), and `PipelineView::from_model`
// renders that. No Spark/SDP type crosses into the viewer — the viewer consumes
// only the normalized model, so a Flink/Beam/skade-streaming adapter would feed
// the SAME `from_model` path. This is the one-viewer-many-adapters seam under test.
fn build_view(vs: &ViewerState) -> PipelineView {
let mut model = knut_pipelines::to_pipeline_model(vs);
model.title = "pipeline".to_string();
PipelineView::from_model(model)
}
// ── the matrix ───────────────────────────────────────────────────────────────
#[test]
fn knut_facett_connector_matrix() {
let shapes = [Shape::Linear, Shape::Diamond, Shape::WideFanOut];
let mixes = [
StatusMix::AllPending,
StatusMix::MidRunMixed,
StatusMix::AllOk,
StatusMix::OneFailed,
];
let progresses = [
ProgressState::NoProgress,
ProgressState::Streaming,
ProgressState::Complete,
];
let mut cells = 0usize;
for &shape in &shapes {
for &mix in &mixes {
for &prog in &progresses {
cells += 1;
let label = format!("{shape:?}/{mix:?}/{prog:?}");
// ── build a REAL knut model + inject REAL run state ──
let graph = shape.graph();
let names: Vec<String> = graph.datasets.iter().map(|d| d.name.clone()).collect();
let mut vs = ViewerState::new(graph);
let failed = mix.apply(&mut vs, &names);
prog.apply(&mut vs, &names);
// ── map through the connector and render headlessly ──
let mut view = build_view(&vs);
let report = facett_core::harness::render_themed(
&mut view,
facett_core::Theme::cyberpunk_neon(),
);
let s = &report.state;
assert!(report.drew(), "[{label}] pipeline drew vertices");
// (1) structure: node + edge counts match the graph.
assert_eq!(
s["node_count"], shape.node_count(),
"[{label}] node count"
);
assert_eq!(
s["edge_count"], shape.edge_count(),
"[{label}] edge count"
);
assert_eq!(
s["nodes"].as_array().unwrap().len(),
shape.node_count(),
"[{label}] per-node array length"
);
// (2) per-node status counts match the injected mix (cross-checked
// against knut's own ViewerState counts — the connector must
// not lose or invent statuses).
let (k_idle, k_run, k_done, k_fail) = vs.status_counts();
assert_eq!(s["statuses"]["idle"], k_idle, "[{label}] idle");
assert_eq!(s["statuses"]["running"], k_run, "[{label}] running");
assert_eq!(s["statuses"]["done"], k_done, "[{label}] done");
assert_eq!(s["statuses"]["failed"], k_fail, "[{label}] failed");
// and the per-mix expectation, spelled out:
match mix {
StatusMix::AllPending => {
assert_eq!(s["statuses"]["idle"], shape.node_count(), "[{label}] all idle");
assert_eq!(s["statuses"]["failed"], 0);
}
StatusMix::AllOk => {
assert_eq!(s["statuses"]["done"], shape.node_count(), "[{label}] all done");
assert_eq!(s["statuses"]["failed"], 0);
}
StatusMix::MidRunMixed => {
assert_eq!(s["statuses"]["done"], 1, "[{label}] one done");
assert_eq!(s["statuses"]["running"], 1, "[{label}] one running");
}
StatusMix::OneFailed => {
assert_eq!(s["statuses"]["failed"], 1, "[{label}] exactly one failed");
}
}
// (3) the failed node's ERROR is surfaced verbatim in errors[].
if let Some((fname, ferr)) = &failed {
let errs = s["errors"].as_array().unwrap();
assert_eq!(errs.len(), 1, "[{label}] one surfaced error");
assert_eq!(errs[0]["node"], fname.as_str(), "[{label}] error attributed to failed node");
assert_eq!(errs[0]["error"], ferr.as_str(), "[{label}] error text reaches the view verbatim");
// and the per-node entry for the failed node carries it too.
let node = s["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| n["name"] == fname.as_str())
.unwrap();
assert_eq!(node["status"], "failed", "[{label}] failed node status");
assert_eq!(node["error"], ferr.as_str(), "[{label}] failed node error");
} else {
assert!(
s["errors"].as_array().unwrap().is_empty(),
"[{label}] no errors when nothing failed"
);
}
// (4) progress numbers match the injected progress (or null).
match prog {
ProgressState::NoProgress => {
assert!(s["progress"].is_null(), "[{label}] no progress block");
}
ProgressState::Streaming => {
let p = &s["progress"];
assert!(!p.is_null(), "[{label}] streaming progress present");
// streaming: 10/16 tasks → fraction 0.625, 6 inflight, 3 MiB.
assert_eq!(p["tasks_total"], 16, "[{label}] total tasks");
assert_eq!(p["tasks_done"], 10, "[{label}] done tasks");
assert_eq!(p["inflight"], 6, "[{label}] inflight");
assert_eq!(p["bytes_read"], 3 * 1024 * 1024, "[{label}] bytes read");
let frac = p["fraction"].as_f64().unwrap();
assert!((frac - 0.625).abs() < 1e-4, "[{label}] fraction {frac}");
// derived rows/s > 0 since elapsed > 0 and rows > 0.
assert!(p["rows_per_s"].as_f64().unwrap() > 0.0, "[{label}] throughput derived");
}
ProgressState::Complete => {
let p = &s["progress"];
assert!(!p.is_null(), "[{label}] complete progress present");
assert_eq!(p["tasks_total"], 10, "[{label}] complete total");
assert_eq!(p["tasks_done"], 10, "[{label}] complete done");
assert_eq!(p["inflight"], 0, "[{label}] none inflight at completion");
let frac = p["fraction"].as_f64().unwrap();
assert!((frac - 1.0).abs() < 1e-4, "[{label}] complete fraction 1.0");
}
}
}
}
}
assert_eq!(cells, 3 * 4 * 3, "the full matrix ran every cell");
eprintln!("knut→facett connector matrix: {cells} cells passed");
// Feed the matrix: ok iff every connector cell rendered + asserted right.
#[cfg(feature = "testmatrix")]
facett_core::testmatrix::emit(
"PipelineView.knut_connector",
"all_cells_pass",
cells == 3 * 4 * 3,
&format!("{cells} of 36 connector cells (shape×status×progress) passed"),
);
}
/// A focused close-up of the richest cell: a diamond, one failed node, mid-stream
/// progress, with a node click-selected — asserting the **selected-node detail**
/// (status + error) reaches `selection_json`, the per-node `detail` (output type)
/// is present, and the failed node's ⚠ error is the selected one. Proves the
/// detail panel surface, not just the aggregate counts.
#[test]
fn selected_failed_node_detail_surfaces() {
let graph = Shape::Diamond.graph();
let names: Vec<String> = graph.datasets.iter().map(|d| d.name.clone()).collect();
let mut vs = ViewerState::new(graph);
let (failed, err) = StatusMix::OneFailed.apply(&mut vs, &names).unwrap();
ProgressState::Streaming.apply(&mut vs, &names);
let view = build_view(&vs).with_selected(&failed);
// selection_json is the click-selected detail the host panel reads.
let sel = view.selection_json();
assert_eq!(sel["node"], failed.as_str(), "selected node is the failed one");
assert_eq!(sel["status"], "failed");
assert_eq!(sel["error"], err.as_str(), "selected node's error text");
// full state: the selected block mirrors it, and detail carries output type.
let s = view.state_json();
assert_eq!(s["selected"]["name"], failed.as_str());
assert_eq!(s["selected"]["status"], "failed");
assert_eq!(s["selected"]["detail"], "materialized_view", "output type as detail line");
assert_eq!(s["selected"]["error"], err.as_str());
// Feed the matrix: ok iff the selected failed node's detail+error surfaced.
#[cfg(feature = "testmatrix")]
facett_core::testmatrix::emit(
"PipelineView.selection_detail",
"failed_node_detail",
sel["node"] == failed.as_str() && sel["error"] == err.as_str(),
&format!("selected failed node '{failed}' surfaced status+error"),
);
}