nordisk/knut

crates/knut-pipelines/src/tests.rs

#![cfg(test)]

use crate::error::PipelineError;
use crate::model::{
    DataflowGraph, Dataset, EventLevel, Flow, OutputType, PipelineRunEvent, SourceLocation,
};
use crate::viewer::{self, PipelineEventSink, StdoutEventSink};

/// A small medallion-style graph: an external ingest into a bronze table, a
/// cleaned silver materialized view, and a gold aggregate view.
fn sample_graph() -> DataflowGraph {
    DataflowGraph {
        graph_id: Some("g-123".into()),
        default_catalog: Some("main".into()),
        default_database: Some("sales".into()),
        ..DataflowGraph::new()
    }
    .with_dataset(Dataset {
        comment: Some("raw orders".into()),
        format: Some("delta".into()),
        ..Dataset::new("bronze_orders", OutputType::Table)
    })
    .with_dataset(Dataset::new("silver_orders", OutputType::MaterializedView))
    .with_dataset(Dataset::new("gold_daily", OutputType::MaterializedView))
    .with_flow(Flow::new("ingest", "bronze_orders", Vec::<String>::new()))
    .with_flow(Flow::new("clean", "silver_orders", ["bronze_orders"]))
    .with_flow(Flow::new("rollup", "gold_daily", ["silver_orders"]))
}

#[test]
fn validate_accepts_well_formed_graph() {
    sample_graph().validate().expect("should be valid");
}

#[test]
fn validate_rejects_dangling_flow() {
    let g = DataflowGraph::new()
        .with_dataset(Dataset::new("a", OutputType::Table))
        .with_flow(Flow::new("f", "missing", ["a"]));
    match g.validate() {
        Err(PipelineError::DanglingFlow { flow, target }) => {
            assert_eq!(flow, "f");
            assert_eq!(target, "missing");
        }
        other => panic!("expected DanglingFlow, got {other:?}"),
    }
}

#[test]
fn validate_rejects_duplicate_dataset() {
    let g = DataflowGraph::new()
        .with_dataset(Dataset::new("a", OutputType::Table))
        .with_dataset(Dataset::new("a", OutputType::MaterializedView));
    assert!(matches!(
        g.validate(),
        Err(PipelineError::DuplicateName {
            kind: "dataset",
            ..
        })
    ));
}

#[test]
fn validate_allows_external_reads() {
    // `external_table` is not a defined dataset — that's fine, it's an
    // upstream source outside the graph.
    let g = DataflowGraph::new()
        .with_dataset(Dataset::new("a", OutputType::Table))
        .with_flow(Flow::new("f", "a", ["external_table"]));
    g.validate().expect("external reads are allowed");
}

#[test]
fn topo_order_follows_flows() {
    let order = sample_graph().topo_order().expect("acyclic");
    assert_eq!(order, vec!["bronze_orders", "silver_orders", "gold_daily"]);
}

#[test]
fn topo_order_detects_cycle() {
    let g = DataflowGraph::new()
        .with_dataset(Dataset::new("a", OutputType::Table))
        .with_dataset(Dataset::new("b", OutputType::Table))
        .with_flow(Flow::new("ab", "b", ["a"]))
        .with_flow(Flow::new("ba", "a", ["b"]));
    assert!(g.topo_order().is_none());
}

#[test]
fn flows_into_filters_by_target() {
    let g = sample_graph();
    let into_silver: Vec<_> = g
        .flows_into("silver_orders")
        .map(|f| f.name.as_str())
        .collect();
    assert_eq!(into_silver, vec!["clean"]);
}

#[test]
fn dump_json_roundtrips() {
    let g = sample_graph();
    let json = viewer::dump_json(&g).unwrap();
    let back: DataflowGraph = serde_json::from_str(&json).unwrap();
    assert_eq!(g, back);
}

#[test]
fn dump_text_mentions_nodes_flows_and_order() {
    let text = viewer::dump_text(&sample_graph());
    assert!(text.contains("Dataflow graph g-123"));
    assert!(text.contains("[table] bronze_orders"));
    assert!(text.contains("(format=delta)"));
    assert!(text.contains("[materialized_view] silver_orders"));
    assert!(text.contains("<- flow clean reads: bronze_orders"));
    assert!(text.contains("(external source)"));
    assert!(text.contains("topological order: bronze_orders -> silver_orders -> gold_daily"));
}

#[test]
fn dump_text_reports_cycle() {
    let g = DataflowGraph::new()
        .with_dataset(Dataset::new("a", OutputType::Table))
        .with_dataset(Dataset::new("b", OutputType::Table))
        .with_flow(Flow::new("ab", "b", ["a"]))
        .with_flow(Flow::new("ba", "a", ["b"]));
    assert!(viewer::dump_text(&g).contains("<cycle detected>"));
}

#[test]
fn source_location_empty_check() {
    assert!(SourceLocation::default().is_empty());
    assert!(
        !SourceLocation {
            line_number: Some(7),
            ..SourceLocation::default()
        }
        .is_empty()
    );
}

#[test]
fn event_sink_formats_and_counts() {
    let events = vec![
        PipelineRunEvent::info("starting run"),
        PipelineRunEvent {
            timestamp: Some("2026-06-10T12:00:00Z".into()),
            level: EventLevel::Warn,
            message: "slow flow".into(),
            element: Some("silver_orders".into()),
        },
        PipelineRunEvent {
            timestamp: None,
            level: EventLevel::Error,
            message: "flow failed".into(),
            element: None,
        },
    ];

    let mut buf: Vec<u8> = Vec::new();
    let mut sink = StdoutEventSink::to_writer(&mut buf);
    for ev in &events {
        sink.on_event(ev).unwrap();
    }
    assert_eq!(sink.count(), 3);
    sink.on_finish().unwrap();

    let out = String::from_utf8(buf).unwrap();
    assert!(out.contains("INFO   [-]  starting run"));
    assert!(out.contains("WARN   [2026-06-10T12:00:00Z]  slow flow  (silver_orders)"));
    assert!(out.contains("ERROR  [-]  flow failed"));
    assert!(out.contains("-- run stream ended (3 event(s)) --"));
}

#[test]
fn format_line_is_single_line() {
    let line = StdoutEventSink::<std::io::Stdout>::format_line(&PipelineRunEvent::info("x"));
    assert!(!line.contains('\n'));
}

#[test]
fn output_type_labels_are_stable() {
    assert_eq!(OutputType::MaterializedView.label(), "materialized_view");
    assert_eq!(OutputType::Sink.label(), "sink");
    assert_eq!(OutputType::Unspecified.label(), "unspecified");
}