diff --git a/src/dispatch.rs b/src/dispatch.rs index 244d184..a019535 100644 --- a/src/dispatch.rs +++ b/src/dispatch.rs @@ -429,6 +429,49 @@ mod tests { assert_eq!(s2.messages_in.load(Ordering::Relaxed), 1); } + #[tokio::test] + async fn forwards_additive_sample_fields_and_explicit_null_value_untouched() { + // SOUTHBOUND §2 ("Explicit null values" + "Additive sample fields"): a SouthboundSignalUpdate + // whose samples carry unknown additive fields (valueType/valueEncoding) and an explicit + // null value with quality GOOD must flow through ingestion without error, without the + // sample being dropped, and without the additive fields being stripped. + let body = json!({ + "signal": { "id": "bacnet-av-3", "name": "AV3" }, + "samples": [{ + "value": null, "quality": "GOOD", "qualityRaw": "relinquished", + "sourceTs": "2026-07-26T00:00:00Z", + "valueType": "REAL", "valueEncoding": "scalar" + }] + }); + let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0") + .southbound_signal_update(body.clone()) + .build(); + + let fake = FakeMessaging::new(); + let messaging: Arc = fake.clone(); + let (_rec, evt) = EvtEmitter::recording(); + let (tx, mut rx) = mpsc::channel(4); + let stats = RouteStats::new("r1"); + + self_subscribe( + &messaging, + "ecv1/+/+/+/data/#", + vec![(tx, stats.clone())], + "my-device".into(), + "telemetry-processor".into(), + evt, + ) + .await + .unwrap(); + + fake.deliver("ecv1/+/+/+/data/#", "ecv1/other/bacnet-adapter/i1/data/av3", m).await; + + let pm = rx.try_recv().expect("the explicit-null sample must be forwarded, not dropped"); + assert_eq!(pm.msg.body, body, "additive sample fields and the null value must survive ingestion verbatim"); + assert_eq!(stats.messages_in.load(Ordering::Relaxed), 1); + assert_eq!(stats.messages_dropped.load(Ordering::Relaxed), 0); + } + #[tokio::test] async fn drops_a_re_consumed_message_carrying_its_own_identity() { let fake = FakeMessaging::new(); diff --git a/src/proc/aggregate.rs b/src/proc/aggregate.rs index 838beff..c543dd9 100644 --- a/src/proc/aggregate.rs +++ b/src/proc/aggregate.rs @@ -315,6 +315,61 @@ mod tests { assert_eq!(agg["count"], json!(5)); } + #[test] + fn explicit_null_samples_fold_per_documented_null_semantics() { + // SOUTHBOUND §2: explicit null values (quality GOOD) must not crash aggregation. This + // repo's documented null semantics (docs/reference/data-types.md "Aggregate `agg` value + // types"): `count` counts ALL folded values (null included), `first`/`last` carry the raw + // value (null included), and numeric reducers are null when the window had no numeric + // sample. Additive sample fields (valueType/...) beside the value must not disturb any of it. + let spec = AggregateSpec { + window: "2".into(), + by: None, + fns: vec!["avg".into(), "sum".into(), "min".into(), "max".into(), "count".into(), "first".into(), "last".into()], + value: None, + }; + let mk = |value: Value, recv: u64| { + let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0") + .payload(json!({ + "signal": { "id": "a" }, + "samples": [ { "value": value, "quality": "GOOD", "qualityRaw": "relinquished", + "valueType": "REAL", "valueEncoding": "scalar" } ] + })) + .build(); + ProcMsg { topic: "t".into(), msg: m, recv_ms: recv } + }; + + // Window 1: all-null. Numeric reducers are null; count still counts both samples. + let mut s = AggregateStage::build(&spec, "body.signal.id").unwrap(); + assert!(s.process(mk(Value::Null, 1)).is_empty()); + let out = s.process(mk(Value::Null, 2)); + assert_eq!(out.len(), 1); + let agg = &out[0].msg.body["agg"]; + assert_eq!(agg["avg"], Value::Null); + assert_eq!(agg["sum"], Value::Null); + assert_eq!(agg["min"], Value::Null); + assert_eq!(agg["max"], Value::Null); + assert_eq!(agg["count"], json!(2)); + assert_eq!(agg["first"], Value::Null); + assert_eq!(agg["last"], Value::Null); + // The primary (first fn = avg) lands as samples[0].value = null without error. + assert_eq!(out[0].msg.body["samples"][0]["value"], Value::Null); + + // Window 2: mixed null + numeric. Numeric reducers fold the numeric values only; count + // includes the null. + assert!(s.process(mk(json!(10.0), 3)).is_empty()); + let out = s.process(mk(Value::Null, 4)); + assert_eq!(out.len(), 1); + let agg = &out[0].msg.body["agg"]; + assert_eq!(agg["avg"], json!(10.0)); + assert_eq!(agg["sum"], json!(10.0)); + assert_eq!(agg["min"], json!(10.0)); + assert_eq!(agg["max"], json!(10.0)); + assert_eq!(agg["count"], json!(2)); + assert_eq!(agg["first"], json!(10.0)); + assert_eq!(agg["last"], Value::Null); + } + #[test] fn aggregate_custom_value_path_non_southbound() { // A non-SouthboundSignalUpdate payload: aggregate `body.temp`, keyed by `body.id`. diff --git a/src/proc/filter.rs b/src/proc/filter.rs index c76d69d..d993fb2 100644 --- a/src/proc/filter.rs +++ b/src/proc/filter.rs @@ -233,6 +233,25 @@ mod tests { assert_eq!(mk("body.signal.id", "contains", Some(json!("zzz"))).process(m()).len(), 0); } + #[test] + fn quality_filter_keeps_explicit_null_value_with_good_quality_and_additive_fields() { + // SOUTHBOUND §2: an explicit null value with quality GOOD is a legitimate sample (e.g. a + // BACnet relinquished slot), and samples MAY carry additive fields (valueType/valueEncoding) + // consumers MUST ignore. A quality filter keys on `quality` only — it must keep the sample. + let spec = FilterSpec { quality: Some("GOOD".into()), ..Default::default() }; + let mut s = FilterStage::build(&spec, ScriptEngineKind::Rhai, &engine(), &ScriptLoader::default(), &ctx()).unwrap(); + let m = msg(json!([{ + "value": null, "quality": "GOOD", "qualityRaw": "relinquished", + "valueType": "REAL", "valueEncoding": "scalar" + }])); + assert_eq!(s.process(m).len(), 1, "explicit-null GOOD sample must not be dropped"); + + // A failed read (null + BAD) is still droppable by the same filter — the two null cases + // stay distinguishable by quality alone. + let bad = msg(json!([{ "value": null, "quality": "BAD", "qualityRaw": "timeout" }])); + assert_eq!(s.process(bad).len(), 0); + } + #[test] fn build_and_op_parse_errors() { // No predicate form configured. diff --git a/src/proc/mod.rs b/src/proc/mod.rs index 5a15126..83be6fe 100644 --- a/src/proc/mod.rs +++ b/src/proc/mod.rs @@ -187,6 +187,50 @@ mod tests { assert_eq!(out[0].msg.body["agg"]["avg"], json!(25.0)); } + #[test] + fn southbound_s2_null_and_additive_sample_fields_flow_through_a_pass_through_pipeline() { + // SOUTHBOUND §2 pin: a representative pass-through pipeline (quality filter → sample → + // project keep) must carry an explicit-null GOOD sample with additive fields + // (valueType/valueEncoding) end to end — no error, no dropped sample, no stripped fields. + use crate::config::{ProjectSpec, StageConfig}; + + let sample = json!({ + "value": null, "quality": "GOOD", "qualityRaw": "relinquished", + "sourceTs": "2026-07-26T00:00:00Z", + "valueType": "REAL", "valueEncoding": "scalar" + }); + let stages = vec![ + StageConfig::Filter(FilterSpec { quality: Some("GOOD".into()), ..Default::default() }), + StageConfig::Sample(SampleSpec { every_n: Some(1), ..Default::default() }), + StageConfig::Project(ProjectSpec { + keep: Some(vec!["signal".into(), "samples".into()]), + set: None, + }), + ]; + let mut p = Pipeline::build( + &stages, + "body.signal.id", + ScriptEngineKind::Rhai, + &engine(), + &script::ScriptLoader::default(), + &Arc::new(script::ScriptContext::default()), + ) + .unwrap(); + + let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0") + .payload(json!({ "signal": { "id": "bacnet-av-3" }, "samples": [sample.clone()] })) + .build(); + let mut input: Out = SmallVec::new(); + input.push(ProcMsg { topic: "t".into(), msg: m, recv_ms: 1 }); + + let out = p.run(input, None); + assert_eq!(out.len(), 1, "the explicit-null GOOD sample must survive the pipeline"); + assert_eq!( + out[0].msg.body["samples"][0], sample, + "additive fields and the null value must pass through verbatim" + ); + } + #[test] fn flush_via_max_tick_closes_open_time_window() { // The `flush` command verb force-closes open TIME windows by running a `u64::MAX` tick diff --git a/src/proc/project.rs b/src/proc/project.rs index 84477ce..62eaff5 100644 --- a/src/proc/project.rs +++ b/src/proc/project.rs @@ -80,6 +80,31 @@ mod tests { assert!(body.get("noise").is_none()); } + #[test] + fn keep_preserves_additive_sample_fields_and_explicit_null_value() { + // SOUTHBOUND §2: samples MAY carry additive fields beside the canonical five, and a + // deliberate null value rides with quality GOOD. `keep` whitelists top-level body keys and + // copies them verbatim — the samples entries (extras and null included) must survive intact. + let sample = json!({ + "value": null, "quality": "GOOD", "qualityRaw": "relinquished", + "sourceTs": "2026-07-26T00:00:00Z", + "valueType": "REAL", "valueEncoding": "scalar" + }); + let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0") + .payload(json!({ "signal": { "id": "t1" }, "samples": [sample.clone()], "noise": 1 })) + .build(); + let pm = ProcMsg { topic: "t".into(), msg: m, recv_ms: now_ms() }; + + let mut s = ProjectStage::build(&ProjectSpec { + keep: Some(vec!["signal".into(), "samples".into()]), + set: None, + }); + let out = s.process(pm); + assert_eq!(out.len(), 1); + assert_eq!(out[0].msg.body["samples"][0], sample, "sample must pass through byte-identical"); + assert!(out[0].msg.body.get("noise").is_none()); + } + #[test] fn set_overlays_literals() { let mut set = Map::new();