facet-wrapped/tests/pipeline_model_wire.rs
//! **Row 5 — the Spark viewer had no way to reach the browser.**
//!
//! `facett-pipeline` already has the seam: `PipelineModel` is the normalized,
//! engine-agnostic shape, `PipelineView::from_model` is the one viewer, and
//! `knut_pipelines::to_pipeline_model` is the Spark/SDP adapter that produces it. The
//! matrix test in `facett-pipeline/tests/knut_facett_matrix.rs` drives exactly that
//! path and says so: *"No Spark/SDP type crosses into the viewer — the viewer consumes
//! only the normalized model."*
//!
//! What was missing is the WIRE. `facet-wrapped`'s `pipeline` adapter took a bespoke
//! `{nodes: [{name, status, rows, …}], edges: [[0,1]]}` shape and built
//! `PipelineNode`s directly, so a `PipelineModel` produced by knut had no JSON
//! encoding that the wasm bundle would accept. The Spark DAG could be normalized and
//! then not travel.
//!
//! ## A correction to how this row was specified
//!
//! The brief says the model can be carried "(serde derives exist)". **They do not.**
//! `PipelineModel`, `Stage`, `StageMetrics` and `PipelineEdge` derive
//! `Clone, Default, Debug` (+`PartialEq`/`Eq` on the edge) and nothing else —
//! `Serialize`/`Deserialize` are on `Status` and `JobProgress`, which is probably
//! where the impression came from. So the encoding is written HERE rather than
//! derived, and it is written in BOTH directions in one place, with a round-trip
//! property test, so an encoder and a decoder cannot drift apart.
//!
//! Row 5 wrote the codec in `facet-wrapped`; row 6 moved it to
//! `facett_pipeline::wire` (still hand-written, still no derives, still both
//! directions in one file) because knut's `knutpipe web` server has to ENCODE it and
//! must not depend on this unpublished, every-component cdylib to do so. The two
//! names imported below are re-exports of the moved codec, so this file drives the
//! same code knut's server now calls.
use facett_pipeline::{
PipelineEdge, PipelineModel, Stage, StageMetrics, StageStatus, UnknownReason,
};
use facet_wrapped::ops::{pipeline_model_from_json, pipeline_model_to_json};
/// A model with every field populated, including the two that are easiest to drop:
/// a distinct `id` (Flink-style opaque ids) and an `Unknown(reason)` status.
fn spark_shaped_model() -> PipelineModel {
PipelineModel {
title: "sdp ingest".into(),
stages: vec![
Stage::new("scan_bronze", StageStatus::Done)
.with_id("stage-0")
.with_kind("table")
.with_metrics(StageMetrics {
rows: Some(1_200_000),
throughput_rows_s: Some(84_000.0),
lag_s: None,
}),
Stage::new("silver_join", StageStatus::Running)
.with_id("stage-1")
.with_kind("materialized_view")
.with_metrics(StageMetrics {
rows: Some(410_000),
throughput_rows_s: Some(12_500.0),
lag_s: Some(3.5),
}),
Stage::new("gold_agg", StageStatus::Failed)
.with_id("stage-2")
.with_error("AnalysisException: cannot resolve `kommun`"),
Stage::new("publish", StageStatus::Unknown(UnknownReason::NotAuthorised))
.with_id("stage-3"),
],
edges: vec![
PipelineEdge::new("stage-0", "stage-1"),
PipelineEdge {
from: "stage-1".into(),
to: "stage-2".into(),
label: Some("12k rows/s".into()),
},
PipelineEdge::new("stage-2", "stage-3"),
],
metrics: None,
}
}
#[test]
fn a_model_survives_the_wire_unchanged() {
// THE row-5 contract, and it is a property rather than a fixture: encode the model
// this crate would receive from knut, decode it, and compare FIELD BY FIELD. A
// stage/edge COUNT check — which is what the brief asked for — would pass on a
// wire that dropped every status, id, kind, error and metric.
let before = spark_shaped_model();
let after = pipeline_model_from_json(&pipeline_model_to_json(&before))
.expect("the wire we just wrote must decode");
assert_eq!(after.title, before.title);
assert_eq!(after.stages.len(), before.stages.len(), "stage count");
assert_eq!(after.edges.len(), before.edges.len(), "edge count");
for (a, b) in after.stages.iter().zip(&before.stages) {
assert_eq!(a.id, b.id, "the stable id is what edges reference");
assert_eq!(a.name, b.name);
assert_eq!(a.status, b.status, "status for `{}`", b.id);
assert_eq!(a.kind, b.kind);
assert_eq!(a.error, b.error);
assert_eq!(a.metrics.rows, b.metrics.rows);
assert_eq!(a.metrics.throughput_rows_s, b.metrics.throughput_rows_s);
assert_eq!(a.metrics.lag_s, b.metrics.lag_s);
}
assert_eq!(after.edges, before.edges, "edges carry their labels");
}
#[test]
fn the_unknown_reason_survives_and_is_not_flattened_to_idle() {
// `StageStatus` documents the distinction as the point of the type: Idle means
// "we know it has not started", Unknown(reason) means "we could not find out".
// A wire that collapsed the second into the first would turn a permissions
// failure into a pipeline that looks merely pending.
let before = spark_shaped_model();
let after = pipeline_model_from_json(&pipeline_model_to_json(&before)).unwrap();
assert_eq!(
after.stages[3].status,
StageStatus::Unknown(UnknownReason::NotAuthorised),
"an unreadable stage came back as {:?}",
after.stages[3].status
);
assert_ne!(after.stages[3].status, StageStatus::Idle);
}
#[test]
fn the_wasm_component_mounts_the_model_shape() {
// The end of the row: the same JSON, through the registry's data seam, produces a
// real pipeline surface — which is what the browser will do with it.
let wire = pipeline_model_to_json(&spark_shaped_model());
let f = facet_wrapped::ops::pipeline(&wire).expect("the model shape must mount");
let st = f.state_json();
let n = st["nodes"].as_array().map(Vec::len).or_else(|| st["node_count"].as_u64().map(|v| v as usize));
assert_eq!(n, Some(4), "state: {st}");
}
#[test]
fn the_registry_accepts_the_model_shape_through_check() {
// `Registry::check` is the pre-flight the loader runs before mounting, so a shape
// the adapter handles but `check` refuses would show the refusal panel in the
// browser while every test here passed.
let wire = pipeline_model_to_json(&spark_shaped_model());
let r = facet_wrapped::Registry::with_builtins();
r.check("pipeline", &wire).expect("check must accept the PipelineModel wire shape");
assert!(r.build_with("pipeline", &wire).is_ok());
}
#[test]
fn the_older_nodes_shape_still_works() {
// knut is not the only caller. The `{nodes, edges:[[0,1]]}` shape is what the
// gallery, the robot and every existing consumer send, and row 5 adds a shape —
// it does not replace one.
let legacy = serde_json::json!({
"title": "ingest",
"nodes": [
{"name": "scan", "status": "done", "rows": 1200},
{"name": "embed", "status": "running", "throughput_rows_s": 840.0},
{"name": "index", "status": "queued"}
],
"edges": [[0, 1], [1, 2]]
});
let f = facet_wrapped::ops::pipeline(&legacy).expect("the legacy shape must still mount");
let st = f.state_json();
let n = st["nodes"].as_array().map(Vec::len).or_else(|| st["node_count"].as_u64().map(|v| v as usize));
assert_eq!(n, Some(3), "state: {st}");
}
#[test]
fn an_empty_or_unrecognisable_payload_is_still_refused() {
// The registry's promise is that a component refuses rather than showing a
// fixture. Adding a second accepted shape must not turn `pipeline` into something
// that accepts anything.
for bad in [
serde_json::json!({}),
serde_json::json!({"title": "x"}),
serde_json::json!({"stages": []}),
serde_json::json!({"nodes": []}),
] {
assert!(
facet_wrapped::ops::pipeline(&bad).is_none(),
"`{bad}` must be refused, not mounted empty"
);
}
}
#[test]
fn edges_naming_a_stage_that_does_not_exist_are_dropped_not_mounted() {
// A Spark plan arriving mid-edit can reference a stage that has not been sent.
// Drawing an edge to nowhere is worse than drawing no edge.
let mut m = spark_shaped_model();
m.edges.push(PipelineEdge::new("stage-0", "stage-does-not-exist"));
let wire = pipeline_model_to_json(&m);
let decoded = pipeline_model_from_json(&wire).unwrap();
assert_eq!(decoded.edges.len(), 4, "the wire carries what it was given");
// ...and the VIEW resolves ids to indices, so the dangling one cannot be drawn.
let f = facet_wrapped::ops::pipeline(&wire).expect("mounts");
let st = f.state_json();
let edges = st["edges"].as_array().map(Vec::len).or_else(|| st["edge_count"].as_u64().map(|v| v as usize));
assert_eq!(edges, Some(3), "the dangling edge must not be drawn: {st}");
}
#[test]
fn the_shipped_gallery_payload_is_the_model_shape_and_keeps_its_unknown_reason() {
// The demo is where a person SEES this row. `gallery_covers_the_manifest` proves
// every payload BUILDS — it does not prove this one is the MODEL shape, and the
// older `nodes` shape builds just as happily.
//
// The `unknown:not_authorised` stage is deliberate in the demo: it is the state a
// Spark plan is actually in when the probe was refused, and it is the one a
// careless wire flattens into "pending".
let table: serde_json::Value = serde_json::from_str(
&std::fs::read_to_string(
std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap()
.join("py/facett_demo/demo_payloads.json"),
)
.expect("the gallery payload table must exist"),
)
.expect("valid JSON");
let wire = &table["payloads"]["pipeline"];
assert!(wire.get("stages").is_some(), "the gallery's pipeline payload is not the model shape");
let m = pipeline_model_from_json(wire).expect("the shipped payload must decode");
assert_eq!(m.stages.len(), 4);
assert_eq!(m.edges.len(), 3);
assert_eq!(m.stages[3].status, StageStatus::Unknown(UnknownReason::NotAuthorised));
assert!(m.stages[2].error.is_some(), "the failed stage keeps its message");
assert_eq!(m.stages[1].metrics.lag_s, Some(3.5), "streaming lag survives");
}