Aura/crates/aura-engine/tests/bind_tap_e2e.rs

//! End-to-end coverage for `FlatGraph::bind_tap` (#282, task 5): the caller-side
//! half of declared taps. `flat_taps_e2e.rs` pins that `compile()` resolves a
//! declared `Tap` into `FlatGraph.taps`; these tests pin the other half — that
//! `bind_tap` actually wires a caller-built sink onto the resolved wire and that
//! it runs through the FULL public pipeline (`Composite::with_taps` ->
//! `compile()` -> `bind_tap` -> `Harness::bootstrap` -> `run`), not just the
//! hand-built `FlatGraph` the in-crate unit test exercises.

use std::sync::mpsc;

use aura_engine::{
    Composite, Firing, Harness, NodeSchema, PortSpec, Role, ScalarKind, Scalar, Tap, TapWire,
    Target, Timestamp, VecSource,
};
use aura_std::{Recorder, Sma};

/// A single-node composite: one root-bound `price` role feeds `Sma(length=1)`
/// (an identity passthrough), declaring `taps` on its one producer (node 0,
/// field 0). No `OutField` is exported — the tap is the composite's only
/// measurement surface, matching the "declared tap on an interior producer
/// with no consuming edge" shape #282 targets.
fn producer(taps: Vec<Tap>) -> Composite {
    Composite::new(
        "producer",
        vec![Sma::builder().bind("length", Scalar::i64(1)).into()],
        vec![],
        vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }],
        vec![],
    )
    .with_taps(taps)
}

/// The `NodeSchema` a `Recorder::new(kinds, firing, tx)` sink satisfies — the
/// caller-side construction contract `bind_tap` requires alongside the boxed
/// node (mirrors `aura_std::Recorder::builder`'s schema, built here directly
/// since `bind_tap` takes a bare `Box<dyn Node>`, not a `PrimitiveBuilder`).
fn recorder_sig(kinds: &[ScalarKind]) -> NodeSchema {
    NodeSchema {
        inputs: kinds
            .iter()
            .enumerate()
            .map(|(i, &kind)| PortSpec { kind, firing: Firing::Any, name: format!("col[{i}]") })
            .collect(),
        output: vec![],
        params: vec![],
        doc: "test-only schema",
    }
}

fn prices(values: &[f64]) -> Vec<(Timestamp, Scalar)> {
    values.iter().enumerate().map(|(i, &v)| (Timestamp(i as i64 + 1), Scalar::f64(v))).collect()
}

/// Property: **`bind_tap` wires a caller-built sink onto the exact flat
/// producer a declared `Tap` resolved to, all the way through the public
/// `Composite::with_taps -> compile() -> bind_tap -> Harness::bootstrap ->
/// run` pipeline.** The recorded trace must be exactly the tapped producer's
/// per-cycle output — not empty (which a wire pointed at the wrong node/field,
/// or a bind that silently no-ops, would also produce for an unrelated
/// column), and not some other node's output.
#[test]
fn bind_tap_records_the_tapped_producers_output_through_the_public_pipeline() {
    let mut flat = producer(vec![Tap { name: "d".into(), from: TapWire { node: 0, field: 0 } }])
        .compile()
        .expect("a tap-bearing composite compiles");

    let (tx, rx) = mpsc::channel();
    flat.bind_tap(
        "d",
        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
        recorder_sig(&[ScalarKind::F64]),
    )
    .expect("the declared tap name binds");

    let mut h = Harness::bootstrap(flat).expect("bootstraps with the appended tap sink");
    h.run(vec![Box::new(VecSource::new(prices(&[10.0, 20.0, 30.0])))]);

    let recorded: Vec<f64> = rx.try_iter().map(|(_, row)| row[0].as_f64()).collect();
    assert_eq!(
        recorded,
        vec![10.0, 20.0, 30.0],
        "the bound tap must record exactly the tapped producer's per-cycle output"
    );
}

/// Property: **`bind_tap` refuses a name the graph did not declare as a tap,
/// and does so without mutating the graph** — the caller can retry with the
/// correct name rather than being left with a half-attached sink or a
/// silently-corrupted node list. `TapBindError` is not part of the crate's
/// public re-export surface (yet), so this asserts the shape observable from
/// outside the crate: an `Err`, and an unmodified flat graph.
#[test]
fn bind_tap_refuses_an_undeclared_tap_name_without_mutating_the_graph() {
    let mut flat = producer(vec![Tap { name: "d".into(), from: TapWire { node: 0, field: 0 } }])
        .compile()
        .expect("a tap-bearing composite compiles");
    let nodes_before = flat.nodes.len();

    let (tx, _rx) = mpsc::channel();
    let result = flat.bind_tap(
        "nope",
        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
        recorder_sig(&[ScalarKind::F64]),
    );

    assert!(result.is_err(), "binding an undeclared tap name must be refused");
    assert_eq!(
        flat.nodes.len(),
        nodes_before,
        "a refused bind must not append the sink to the flat node list"
    );
}