nordisk/knut

crates/knut-thund/tests/testmatrix_wiring.rs

//! Wiring proof for knut-thund's `functional_status` test-matrix surface (the
//! newest feature: the IR validation gate and the IR→SDP lowering report their
//! health into the nornir matrix; the execution backends report their honest
//! outcome, or — built without their feature — the capability gap).
//!
//! INJECT-AND-ASSERT (project LAW): the IR surfaces are pure/offline, so this
//! drives the REAL `Pipeline::validate` + `Pipeline::to_sdp` over real
//! pipelines and asserts on the REAL emitted rows. The native/spark execution
//! backends are feature-gated (heavy DataFusion / Spark Connect trees) and not
//! built here, so the RED capability-gap row their `run` emits without its
//! feature is asserted through the same public wrapper. Offline, in-process.
//!
//! One test per build (single test per binary → no race on the process-global
//! functional buffer `--features testmatrix` records into):
//!   * default build → the emit is a compiled-out no-op; assert the wrapper
//!     links and the real IR surfaces still work.
//!   * `--features testmatrix` → assert `validate` records green/red, `to_sdp`
//!     records green, and the backend capability-gap row records RED (the
//!     engine is not compiled in without `--features native`/`spark`).

use knut_thund::ir::{Dataset, Flow, OutputType, Pipeline};

/// A small, valid pipeline: one dataset fed by one batch flow.
fn good_pipeline() -> Pipeline {
    Pipeline::new("daily_sales")
        .with_dataset(Dataset::new("clean_orders", OutputType::MaterializedView))
        .with_flow(Flow::batch("f_clean", "clean_orders", Vec::<String>::new()))
}

/// An invalid pipeline: the flow targets a dataset that does not exist.
fn bad_pipeline() -> Pipeline {
    Pipeline::new("broken").with_flow(Flow::batch("f", "ghost", Vec::<String>::new()))
}

/// Default build: `testmatrix` OFF → `functional_status` is an inlined no-op.
/// Prove it links and the IR surfaces it is wired into still work.
#[cfg(not(feature = "testmatrix"))]
#[test]
fn functional_status_is_a_linked_noop_and_ir_surfaces_run() {
    knut_thund::functional_status("knut-thund/ir", "validate", true, "noop");

    good_pipeline().validate().expect("valid pipeline");
    assert!(
        bad_pipeline().validate().is_err(),
        "dangling flow is rejected"
    );

    let sdp = good_pipeline().to_sdp();
    assert_eq!(sdp.datasets.len(), 1, "lowering preserves the dataset");
}

/// `--features testmatrix`: `validate` records green for a valid pipeline and
/// red for an invalid one; `to_sdp` records green; and the execution-backend
/// scaffolds record honest RED rows (never green).
#[cfg(feature = "testmatrix")]
#[test]
fn ir_surfaces_record_health_and_scaffolds_stay_red() {
    use nornir_testmatrix::{drain_functional_rows, status};

    let _ = drain_functional_rows();

    // Valid pipeline → green validate row + green lower_to_sdp row.
    good_pipeline().validate().expect("valid");
    let _ = good_pipeline().to_sdp();

    // Invalid pipeline → red validate row.
    assert!(bad_pipeline().validate().is_err());

    // Mirror the native execution backend's honest emit when its engine is not
    // compiled in (the `native` DataFusion tree is a heavy feature not built
    // here): a RED `execute` row noting the capability gap. With `--features
    // native` this same row goes GREEN once the batch dataflow actually runs.
    knut_thund::functional_status(
        "knut-thund/backend_native",
        "execute",
        false,
        "native feature disabled: DataFusion engine not compiled in",
    );

    let rows = drain_functional_rows();

    let validates: Vec<_> = rows
        .iter()
        .filter(|r| r.suite == "knut-thund/ir" && r.test_name == "validate")
        .collect();
    assert_eq!(validates.len(), 2, "two validate rows; got {rows:?}");
    assert!(
        status::is_green(&validates[0].status),
        "valid pipeline is GREEN: {:?}",
        validates[0]
    );
    assert_eq!(validates[0].aspect, "functional");
    assert_eq!(
        validates[0].message, "daily_sales",
        "green detail carries the pipeline name"
    );
    assert!(
        status::is_red(&validates[1].status),
        "invalid pipeline is RED: {:?}",
        validates[1]
    );

    let lower = rows
        .iter()
        .find(|r| r.suite == "knut-thund/ir" && r.test_name == "lower_to_sdp")
        .unwrap_or_else(|| panic!("lowering surface must record; got {rows:?}"));
    assert!(
        status::is_green(&lower.status),
        "lowering is GREEN: {lower:?}"
    );
    assert!(
        lower.message.contains("1 dataset(s)"),
        "detail carries counts: {lower:?}"
    );

    let scaffold = rows
        .iter()
        .find(|r| r.suite == "knut-thund/backend_native" && r.test_name == "execute")
        .unwrap_or_else(|| panic!("native scaffold must record; got {rows:?}"));
    assert!(
        status::is_red(&scaffold.status),
        "the native execution backend is a scaffold → RED, never green: {scaffold:?}"
    );
}