using Google.Protobuf; using Org.Eclipse.Tahu.Protobuf; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; /// /// End-to-end ingest of flattened structured metrics (Gitea #515): a tag authored against /// Motor/Speed is fed by a DBIRTH/DDATA carrying a Template metric named /// Motor. /// /// /// pins the mapping in isolation; this file pins that the /// chain actually connects — flatten → alias-table binding → authored-tag lookup → publish. Those /// are four separate places a member name is composed, and a mismatch in any one of them produces /// a driver that decodes structured metrics perfectly and feeds nothing. /// [Trait("Category", "Unit")] public sealed class SparkplugStructuredIngestTests { private const string Group = "Plant1"; private const string Node = "EdgeA"; private const string Device = "Filler1"; private const ulong MotorAlias = 20UL; private const ulong TableAlias = 21UL; private const uint StatusGood = 0x00000000u; private const string SpeedPath = "Plant/Mqtt/spb/MotorSpeed"; private const string TempCellPath = "Plant/Mqtt/spb/TempCell0"; [Fact] public async Task A_template_member_feeds_the_tag_authored_against_its_flattened_name() { // The whole point of #515, asserted end to end. `Motor` is what the wire carries; `Motor/Speed` // is what an operator authors, and nothing on the wire is ever called that. var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); fired.ShouldContainKey(SpeedPath); fired[SpeedPath].Value.ShouldBe(1500d); fired[SpeedPath].StatusCode.ShouldBe(StatusGood); } [Fact] public async Task A_ddata_template_carrying_only_the_parents_alias_still_reaches_the_member_tag() { // The load-bearing case, and the one a naive implementation gets wrong. A DDATA metric carries // an ALIAS and no name — so the member names cannot be composed until the alias table has // resolved the parent's name. Flattening at decode time alone is not enough. var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); fired.Clear(); // seq 2, alias only — no Name on the metric at all. var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL }; data.Metrics.Add(new Payload.Types.Metric { Alias = MotorAlias, TemplateValue = TemplateOf(speed: 1750, running: false), }); ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data)); fired.ShouldContainKey(SpeedPath); fired[SpeedPath].Value.ShouldBe(1750d); } [Fact] public async Task A_dataset_cell_feeds_the_tag_authored_against_its_indexed_name() { var (ing, fired) = await NewIngestorAsync(Tag(TempCellPath, Blob("Table/Temp/0", "Float64"))); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, TableDataSetMetric(20.5, 21.5))); fired.ShouldContainKey(TempCellPath); fired[TempCellPath].Value.ShouldBe(20.5d); } [Fact] public async Task A_member_that_carries_no_datatype_falls_back_to_the_birth_binding() { // A DDATA template is not obliged to re-declare its members' datatypes. Guessing from the CLR // type of the wire value would read a signed Int32 of -42 out of its unsigned field as // 4294967254 — plausible, Good quality, and wrong. var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Offset", "Int32"))); var birthTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false }; birthTemplate.Metrics.Add(new Payload.Types.Metric { Name = "Offset", Datatype = (uint)TahuDataType.Int32, IntValue = unchecked((uint)-42), }); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric { Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = birthTemplate, })); fired[SpeedPath].Value.ShouldBe(-42); fired.Clear(); // DDATA member with NO datatype of its own. var dataTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false }; dataTemplate.Metrics.Add(new Payload.Types.Metric { Name = "Offset", IntValue = unchecked((uint)-7) }); var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL }; data.Metrics.Add(new Payload.Types.Metric { Alias = MotorAlias, TemplateValue = dataTemplate }); ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data)); // -7, not 4294967289: the birth's Int32 binding drove ReinterpretSigned. fired[SpeedPath].Value.ShouldBe(-7); } [Fact] public async Task A_tag_authored_against_the_structured_parent_itself_is_never_fed() { // The parent has no value — its members do. Feeding it anything would require choosing which // member "the" value is, which is exactly the guess #515 refuses to make. var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor", "Float64"))); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); fired.ShouldNotContainKey(SpeedPath); } [Fact] public async Task An_unflattenable_structured_metric_still_feeds_nothing_rather_than_a_guess() { // Skip-and-warn survives — the explicit requirement on the issue. A template DEFINITION is the // realistic instance of this: it names members and carries no values. var definition = new Payload.Types.Template { Version = "1", IsDefinition = true }; definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)TahuDataType.Double }); var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric { Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = definition, })); fired.ShouldNotContainKey(SpeedPath); } [Fact] public async Task The_rediscovery_signature_tracks_members_not_the_parent_metric_name() { // A UDT gaining or losing a member leaves the PARENT metric's name identical, so a gate built // on raw wire names would never move — and the operator would never be told the tree changed, // which for a UDT is the change that matters most. var (ing, _) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); var observed = new List>(); ing.BirthObserved += (_, e) => observed.Add(e.MetricNames); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1, running: true))); var deviceBirth = observed[^1]; deviceBirth.ShouldBe(["Motor/Speed", "Motor/Running"], ignoreOrder: true); deviceBirth.ShouldNotContain("Motor"); } // ---- helpers --------------------------------------------------------------------------------- /// /// A registered and subscribed ingestor. The subscribe step is not optional scaffolding: /// SparkplugIngestor.Raise only fires OnDataChange for a RawPath that holds a /// subscription handle, so an unsubscribed ingestor caches values silently and every assertion /// in this file would read as "no value" whether or not the flattening worked. /// private static async Task<(SparkplugIngestor Ingestor, Dictionary Fired)> NewIngestorAsync( params RawTagEntry[] tags) { var options = new MqttDriverOptions { Mode = MqttMode.SparkplugB, Sparkplug = new MqttSparkplugOptions { GroupId = Group, RequestRebirthOnGap = false }, RawTags = tags, }; var ingestor = new SparkplugIngestor( options, "drv-1", subscribeTransport: new GrantingSubscribeTransport(), publishTransport: null, logger: null, cache: new LastValueCache()); ingestor.Register(tags); await ingestor.SubscribeAsync( [.. ingestor.AuthoredRawPaths], TimeSpan.Zero, TestContext.Current.CancellationToken); var fired = new Dictionary(StringComparer.Ordinal); ingestor.OnDataChange += (_, e) => fired[e.FullReference] = e.Snapshot; return (ingestor, fired); } /// A SUBACK that grants everything; the tests here are about values, not subscribe policy. private sealed class GrantingSubscribeTransport : IMqttSubscribeTransport { public Task> SubscribeAsync( IReadOnlyList filters, CancellationToken cancellationToken) => Task.FromResult>( [.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, true, "GrantedQoS1"))]); } private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false); private static string Blob(string metric, string dataType) => $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"{{metric}}","dataType":"{{dataType}}"}"""; private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) => new(type, Group, Node, device, null); private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray()); private static SparkplugPayload Nbirth() { var payload = new Payload { Seq = 0UL, Timestamp = 1721822400000UL }; payload.Metrics.Add(new Payload.Types.Metric { Name = "bdSeq", Datatype = (uint)TahuDataType.Uint64, LongValue = 1UL, }); return Decode(payload); } private static SparkplugPayload Birth(ulong seq, params Payload.Types.Metric[] metrics) { var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL }; payload.Metrics.AddRange(metrics); return Decode(payload); } private static Payload.Types.Template TemplateOf(double speed, bool running) { var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; template.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)TahuDataType.Double, DoubleValue = speed, }); template.Metrics.Add(new Payload.Types.Metric { Name = "Running", Datatype = (uint)TahuDataType.Boolean, BooleanValue = running, }); return template; } private static Payload.Types.Metric MotorTemplateMetric(double speed, bool running) => new() { Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = TemplateOf(speed, running), }; private static Payload.Types.Metric TableDataSetMetric(params double[] temps) { var dataSet = new Payload.Types.DataSet { NumOfColumns = 1UL }; dataSet.Columns.Add("Temp"); dataSet.Types_.Add((uint)TahuDataType.Double); foreach (var temp in temps) { var row = new Payload.Types.DataSet.Types.Row(); row.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = temp }); dataSet.Rows.Add(row); } return new Payload.Types.Metric { Name = "Table", Alias = TableAlias, Datatype = (uint)TahuDataType.DataSet, DatasetValue = dataSet, }; } }