crates/knut-pipelines/tests/robot_snarl_gui.rs
//! **Robot UI test for the knut-pipelines snarl viewer** (feature `gui`).
//!
//! This is the pipelines-GUI analogue of `facett-demo/tests/robot_demo.rs`:
//! fire up the real egui/`egui-snarl` viewer **headless**, load the **built-in
//! demo graph**, drive a **synthetic, no-server** SDP run through it (the same
//! `ViewerUpdate`s a live Spark Connect stream would deliver, minus the wire),
//! render a **real egui frame** after each step, and assert the viewer
//! **responded** — via the introspection `state_json` **and** the live status
//! carried by each **snarl node** in the DAG. Every `(drive → expected)` step is
//! one row; the run gates on every row being green + a substantial step count
//! (the drive+assert+coverage-gate spirit, adapted to the egui-0.28 stack).
//!
//! ## Why no `egui_kittest`
//! `egui-snarl` 0.4 pins the GUI to **egui 0.28**, and `egui_kittest` does not
//! exist for 0.28 — that crate's first release is **0.30**, and two egui majors
//! can't share one harness. So instead of AccessKit label-clicking, the robot
//! runs frames through `egui::Context::run_ui` directly (the exact headless
//! frame-run kittest wraps internally) and uses `state_json` + the snarl node
//! colours as the post-condition oracle. No window, no display, no Spark server.
//!
//! Run: `cargo test -p knut-pipelines --features gui --test robot_snarl_gui -- --nocapture`
#![cfg(feature = "gui")]
use knut_pipelines::egui;
use knut_pipelines::gui::ViewerApp;
use knut_pipelines::{
EventLevel, ExecutionProgress, NodeStatus, PipelineRunEvent, StageProgress, ViewerState,
ViewerUpdate, demo_graph,
};
use serde_json::{Value, json};
/// One driven step's outcome — a red/green row (the matrix cell).
struct Row {
name: String,
ok: bool,
msg: String,
}
/// A synthetic pipeline event (the shape an SDP flow-progress line takes); its
/// dataset is recovered from the quoted flow name in the message (no `element`),
/// exactly as a real run's free-text events are attributed.
fn ev(msg: &str) -> ViewerUpdate {
ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: msg.into(),
element: None,
})
}
/// A headless robot over the real [`ViewerApp`]: it applies synthetic updates,
/// runs true egui frames, and records assertions as red/green rows.
struct Robot {
app: ViewerApp,
ctx: egui::Context,
rows: Vec<Row>,
}
impl Robot {
/// Build the viewer over the built-in demo graph (no connect endpoint) and
/// land one frame so the panels + snarl DAG lay out before we query them.
fn new() -> Self {
let app = ViewerApp::new(ViewerState::new(demo_graph()), None);
let mut r = Robot {
app,
ctx: egui::Context::default(),
rows: Vec::new(),
};
r.frame();
r
}
/// Render one real egui frame through the viewer's true `draw` path.
fn frame(&mut self) {
let ctx = self.ctx.clone();
let app = &mut self.app;
let mut input = egui::RawInput::default();
input.screen_rect = Some(egui::Rect::from_min_size(
egui::pos2(0.0, 0.0),
egui::vec2(1280.0, 800.0),
));
ctx.run_ui(input, |ui| app.draw(ui)).drop_without_applying_deltas();
}
/// Apply one synthetic update, then render a frame — the robot's "do a thing".
fn drive(&mut self, update: ViewerUpdate) {
self.app.apply_update(update);
self.frame();
}
fn state(&self) -> Value {
self.app.app_state_json()
}
/// Record a red/green row.
fn check(&mut self, name: &str, ok: bool, msg: impl Into<String>) {
self.rows.push(Row {
name: name.to_string(),
ok,
msg: msg.into(),
});
}
/// Assert a `state_json` pointer equals `expected`; record the row.
fn check_state(&mut self, name: &str, pointer: &str, expected: Value) {
let actual = self.state().pointer(pointer).cloned();
let ok = actual.as_ref() == Some(&expected);
let msg = match actual {
Some(_) if ok => String::new(),
Some(a) => format!("at {pointer}: expected {expected}, got {a}"),
None => format!("at {pointer}: absent"),
};
self.check(name, ok, msg);
}
/// Assert the **snarl node** for `dataset` carries `expected` status; row.
fn check_snarl(&mut self, name: &str, dataset: &str, expected: NodeStatus) {
let got = self.app.snarl_status_of(dataset);
let ok = got == Some(expected);
self.check(
name,
ok,
if ok {
String::new()
} else {
format!("snarl node {dataset:?}: expected {expected:?}, got {got:?}")
},
);
}
}
#[test]
fn robot_drives_snarl_viewer_to_all_done() {
let mut r = Robot::new();
// ── 0. The built-in demo graph loaded and the snarl DAG drew. ──
r.check(
"snarl drew one node per dataset (3)",
r.app.snarl_node_count() == 3,
format!("nodes = {}", r.app.snarl_node_count()),
);
r.check_state("demo graph carries 3 datasets", "/dataset_count", json!(3));
r.check_state("demo graph carries 3 flows", "/flow_count", json!(3));
r.check_state(
"datasets in graph order",
"/datasets",
json!(["bronze_events", "silver_events", "gold_daily"]),
);
r.check_state(
"topo order bronze → silver → gold",
"/topo_order",
json!(["bronze_events", "silver_events", "gold_daily"]),
);
r.check_state(
"all three nodes idle at start",
"/status_counts/idle",
json!(3),
);
r.check_state(
"static viewer reports the no-server source",
"/source",
json!("static (no server)"),
);
{
let s = r.state();
let ph: Vec<String> = s["placeholders"]
.as_array()
.unwrap()
.iter()
.map(|p| p.as_str().unwrap().to_string())
.collect();
r.check(
"empty-run shows the `no run events yet` placeholder",
ph.iter().any(|p| p == "no run events yet"),
format!("placeholders = {ph:?}"),
);
}
r.check_snarl(
"snarl bronze node starts idle",
"bronze_events",
NodeStatus::Idle,
);
// ── 1. ingest STARTING → bronze_events goes Running (state + snarl node). ──
r.drive(ev("Flow 'ingest' is STARTING."));
r.check_state(
"ingest starting → bronze_events running",
"/node_status/bronze_events",
json!("running"),
);
r.check_snarl(
"snarl bronze node turns Running",
"bronze_events",
NodeStatus::Running,
);
r.check_state("one node running", "/status_counts/running", json!(1));
// ── 2. ingest COMPLETED (wrote 1,000 rows) → bronze Done + rows attributed. ──
r.drive(ev("Flow 'ingest' has COMPLETED and wrote 1,000 rows."));
r.check_state(
"ingest completed → bronze_events done",
"/node_status/bronze_events",
json!("done"),
);
r.check_snarl(
"snarl bronze node turns Done",
"bronze_events",
NodeStatus::Done,
);
r.check_state(
"bronze row count attributed (1000)",
"/rows_written/bronze_events",
json!(1000),
);
r.check_state(
"last_event reflects the completion line",
"/last_event",
json!("Flow 'ingest' has COMPLETED and wrote 1,000 rows."),
);
// ── 3. clean RUNNING → silver_events Running. ──
r.drive(ev("Flow 'clean' is RUNNING."));
r.check_state(
"clean running → silver_events running",
"/node_status/silver_events",
json!("running"),
);
r.check_snarl(
"snarl silver node turns Running",
"silver_events",
NodeStatus::Running,
);
// ── 4. A Spark job-progress snapshot lands in the side panel. ──
r.drive(ViewerUpdate::Progress(ExecutionProgress {
stages: vec![
StageProgress {
stage_id: 1,
num_tasks: 8,
num_completed_tasks: 8,
input_bytes_read: 2_097_152,
done: true,
},
StageProgress {
stage_id: 2,
num_tasks: 8,
num_completed_tasks: 2,
input_bytes_read: 524_288,
done: false,
},
],
num_inflight_tasks: 6,
}));
r.check_state(
"job progress: 16 total tasks",
"/job_progress/total_tasks",
json!(16),
);
r.check_state(
"job progress: 10 completed tasks",
"/job_progress/completed_tasks",
json!(10),
);
r.check_state(
"job progress not all-done mid-run",
"/job_progress/all_done",
json!(false),
);
// ── 5. clean COMPLETED (500 rows) → silver Done. ──
r.drive(ev("Flow 'clean' has COMPLETED and wrote 500 rows."));
r.check_state(
"clean completed → silver_events done",
"/node_status/silver_events",
json!("done"),
);
r.check_snarl(
"snarl silver node turns Done",
"silver_events",
NodeStatus::Done,
);
r.check_state(
"silver row count attributed (500)",
"/rows_written/silver_events",
json!(500),
);
// ── 6. rollup COMPLETED (50 rows) → gold Done; the whole DAG is done. ──
r.drive(ev("Flow 'rollup' has COMPLETED and wrote 50 rows."));
r.check_state(
"rollup completed → gold_daily done",
"/node_status/gold_daily",
json!("done"),
);
r.check_snarl("snarl gold node turns Done", "gold_daily", NodeStatus::Done);
r.check_state("all three nodes done", "/status_counts/done", json!(3));
r.check_state("no node left idle", "/status_counts/idle", json!(0));
{
let s = r.state();
let ph = s["placeholders"].as_array().unwrap().len();
r.check(
"no placeholders once the run has data",
ph == 0,
format!("placeholders left = {ph}"),
);
}
// ── Coverage gate: every dataset node in the snarl DAG was driven to Done. ──
for ds in ["bronze_events", "silver_events", "gold_daily"] {
r.check_snarl(
&format!("coverage: snarl {ds} reached Done"),
ds,
NodeStatus::Done,
);
}
// ── Print the driven matrix + gate. ──
let passed = r.rows.iter().filter(|r| r.ok).count();
let total = r.rows.len();
eprintln!("\n══ knut-pipelines SNARL VIEWER ROBOT DRIVE (headless egui 0.28) ══");
eprintln!("driving the built-in demo DAG through a synthetic no-server SDP run");
for row in &r.rows {
let mark = if row.ok { "✓" } else { "✗" };
eprintln!(
" {mark} {}{}",
row.name,
if row.msg.is_empty() {
String::new()
} else {
format!(" — {}", row.msg)
}
);
}
eprintln!("── {passed}/{total} robot steps PASSED ──\n");
let reds: Vec<_> = r
.rows
.iter()
.filter(|r| !r.ok)
.map(|r| (&r.name, &r.msg))
.collect();
assert!(
reds.is_empty(),
"snarl-viewer robot has red cells: {reds:#?}"
);
assert!(
total >= 25,
"expected a substantial live-drive scenario, got {total} steps"
);
}
/// Port fidelity: prove the runner records a **wrong** expectation as a red row
/// (not a panic, not a silent pass) — so the gate above is genuinely
/// assertion-driven and never masks a real failure.
#[test]
fn robot_records_a_wrong_expectation_as_a_red_row() {
let mut r = Robot::new();
// A correct assertion is green.
r.check_state(
"good: demo graph has 3 datasets",
"/dataset_count",
json!(3),
);
// A wrong one is red (the graph has 3, not 99), recorded with the reason.
r.check_state(
"bad: dataset_count == 99 (wrong)",
"/dataset_count",
json!(99),
);
// An absent pointer is red too.
r.check_state("bad: nonexistent pointer", "/does/not/exist", json!(1));
assert!(r.rows[0].ok, "the real assertion passed");
assert!(
!r.rows[1].ok,
"a wrong expectation is a red row, not a pass"
);
assert!(
r.rows[1].msg.contains("99"),
"the red row records why: {}",
r.rows[1].msg
);
assert!(!r.rows[2].ok, "an absent pointer is a red row");
}