nordisk/nornir

tests/funnel_demo.rs

//! E1 — funnel **demo injector** golden test (inject-and-assert, data-only).
//!
//! [`inject_demo`] into a tempdir warehouse, REOPEN (forces a full replay of
//! the Iceberg `funnel_events` table → proves durability, not just in-memory
//! state), then assert the **exact** expected nodes/edges/ready-set computed
//! independently from the same deterministic spec. No "didn't panic" — every
//! count and the ready frontier is asserted against a recomputed golden.

use std::collections::{BTreeMap, BTreeSet};

use nornir::funnel::demo::{DEMO_SOURCE, build_dep_graph_spec, inject_demo};
use nornir::funnel::{NodeStatus, Store};

#[test]
fn inject_demo_roundtrips_to_exact_golden_dag() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("wh");

    // --- inject size=3 (→ 1 plan, 9 nodes, 3 layers, seed 0xF1) ---
    let (idea_id, plan_ids) = {
        let mut store = Store::open(&root).unwrap();
        let injected = inject_demo(&mut store, 3).unwrap();
        injected.expect("first inject must write")
    };
    assert_eq!(plan_ids.len(), 1, "inject_demo makes exactly one plan");

    // The independently-recomputed golden spec for size=3 (matches inject_demo).
    let spec = build_dep_graph_spec(1, 3 * 3, 3, 0xF1);
    let plan_spec = &spec[0];
    let expected_nodes = plan_spec.len();
    let expected_edges: usize = plan_spec.iter().map(|s| s.deps.len()).sum();
    assert_eq!(expected_nodes, 9, "size 3 → 9 nodes");
    assert!(expected_edges > 0, "a layered DAG must have edges");

    // Expected ready set: nodes with NO deps are Ready after promote_ready
    // (their deps — none — are vacuously Done); everything with deps is Pending
    // until a dep is Done (nothing is Done in a fresh demo).
    let expected_ready: BTreeSet<String> = plan_spec
        .iter()
        .filter(|s| s.deps.is_empty())
        .map(|s| s.node_id.to_string())
        .collect();
    assert!(!expected_ready.is_empty(), "at least one root → ready");

    // --- REOPEN: full replay from disk ---
    let mut store = Store::open(&root).unwrap();

    // The demo idea is present and tagged for idempotency.
    assert_eq!(
        store.funnel.ideas.get(&idea_id).map(|i| i.source.as_str()),
        Some(DEMO_SOURCE),
        "demo idea persisted with the idempotency tag"
    );

    let plan = store
        .funnel
        .plans
        .get(&plan_ids[0])
        .expect("plan persisted across reopen");
    assert_eq!(plan.nodes.len(), expected_nodes, "node count golden");
    assert_eq!(plan.edges.len(), expected_edges, "edge count golden");

    // Exact edge set golden: every (from -> to) the spec declares is present.
    let expected_edge_set: BTreeSet<(String, String)> = plan_spec
        .iter()
        .flat_map(|s| {
            s.deps
                .iter()
                .map(move |d| (d.to_string(), s.node_id.to_string()))
        })
        .collect();
    let actual_edge_set: BTreeSet<(String, String)> = plan
        .edges
        .iter()
        .map(|(f, t)| (f.to_string(), t.to_string()))
        .collect();
    assert_eq!(actual_edge_set, expected_edge_set, "exact edge set golden");

    // Ready frontier golden after promote_ready.
    store.funnel.promote_ready();
    let plan = store.funnel.plans.get(&plan_ids[0]).unwrap();
    let actual_ready: BTreeSet<String> = plan
        .nodes
        .values()
        .filter(|n| n.status == NodeStatus::Ready)
        .map(|n| n.id.to_string())
        .collect();
    assert_eq!(actual_ready, expected_ready, "ready frontier golden");

    // Every non-ready node is Pending (nothing Done yet) — assert the full
    // status map shape, not just the ready set.
    let mut status: BTreeMap<String, NodeStatus> = BTreeMap::new();
    for n in plan.nodes.values() {
        status.insert(n.id.to_string(), n.status);
    }
    for (id, st) in &status {
        if expected_ready.contains(id) {
            assert_eq!(*st, NodeStatus::Ready, "{id} should be Ready");
        } else {
            assert_eq!(*st, NodeStatus::Pending, "{id} should be Pending");
        }
    }
}

#[test]
fn inject_demo_is_idempotent() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("wh");

    {
        let mut store = Store::open(&root).unwrap();
        assert!(
            inject_demo(&mut store, 2).unwrap().is_some(),
            "first inject writes"
        );
        let plans_after_first = store.funnel.plans.len();
        // Second inject on the SAME store: the demo idea exists → no-op.
        assert!(
            inject_demo(&mut store, 2).unwrap().is_none(),
            "second inject is a no-op"
        );
        assert_eq!(
            store.funnel.plans.len(),
            plans_after_first,
            "no duplicate plan on rerun"
        );
    }

    // And across a reopen the dedup still holds (it keys off the persisted idea).
    let mut store = Store::open(&root).unwrap();
    assert!(
        inject_demo(&mut store, 2).unwrap().is_none(),
        "rerun after reopen is still a no-op"
    );
}