using System.Buffers.Binary; 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; /// /// lmxopcua#754 (R06:M-5): Sparkplug B *Array metrics are read into 1-D OPC UA array values. Every /// test builds real Tahu protobuf payloads, encodes them to wire bytes, decodes them through the real /// and feeds — the method MQTTnet's /// dispatcher thread reaches — so this is the driver's production ingest path end to end, short of a broker. /// (The project's MiniBroker never forwards a PUBLISH, so there is no in-process broker to route /// through; the live Mosquitto + simulator suite is Driver.Mqtt.IntegrationTests.) /// public sealed class SparkplugArrayIngestTests { private const uint Good = 0x00000000u; private const uint BadDecodingError = 0x80070000u; private const uint BadTypeMismatch = 0x80740000u; private const string Group = "Plant1"; private const string Node = "EdgeA"; private const string Device = "Filler1"; private const string ArrPath = "Plant/Mqtt/spb/Profile"; private const ulong ArrAlias = 11UL; // ---- the value: a 1-D array of the tag's element type ---- /// An unauthored dataType takes the birth's ELEMENT type, and the value is an int[] — not a byte[]. [Fact] public async Task Int32Array_OnAnArrayTag_PublishesAnInt32Array_AtGood() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Birth(ing, TahuDataType.Int32Array, Int32s(-42, 0, 70000)); var snapshot = seen.ShouldHaveSingleItem(); snapshot.StatusCode.ShouldBe(Good); snapshot.Value.ShouldBeOfType().ShouldBe(new[] { -42, 0, 70000 }); } /// The DATA path (alias only, no datatype on the wire) decodes by the birth's binding, like a scalar. [Fact] public async Task Ddata_ByAlias_PublishesTheNewVector() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Birth(ing, TahuDataType.Int32Array, Int32s(1, 2)); var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL }; data.Metrics.Add(new Payload.Types.Metric { Alias = ArrAlias, BytesValue = ByteString.CopyFrom(Int32s(3, 4, 5)) }); ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data)); seen.Count.ShouldBe(2); seen[1].StatusCode.ShouldBe(Good); seen[1].Value.ShouldBe(new[] { 3, 4, 5 }); ing.Values.Read(ArrPath).Value.ShouldBe(new[] { 3, 4, 5 }); } /// An authored element type wins over the birth's, element by element through the scalar coercion. [Fact] public async Task AnAuthoredElementType_Wins_OverTheBirth() { var (ing, seen) = await SubscribedAsync(ArrayBlob(dataType: "Float64")); Birth(ing, TahuDataType.Int32Array, Int32s(-1, 2)); seen.ShouldHaveSingleItem().Value.ShouldBeOfType().ShouldBe(new[] { -1d, 2d }); } /// Int8 widens to Int16, exactly as the scalar map does (no OPC UA SByte in DriverDataType). [Fact] public async Task Int8Array_WidensToAnInt16Array() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Birth(ing, TahuDataType.Int8Array, [0xE9, 0x7B]); seen.ShouldHaveSingleItem().Value.ShouldBeOfType().ShouldBe(new short[] { -23, 123 }); } [Fact] public async Task BooleanArray_PublishesABoolArray() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Birth(ing, TahuDataType.BooleanArray, [0x03, 0x00, 0x00, 0x00, 0xA0]); seen.ShouldHaveSingleItem().Value.ShouldBeOfType().ShouldBe(new[] { true, false, true }); } [Fact] public async Task StringArray_PublishesAStringArray() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Birth(ing, TahuDataType.StringArray, [0x41, 0x42, 0x43, 0x00, 0x00, 0x7A, 0x00]); seen.ShouldHaveSingleItem().Value.ShouldBeOfType().ShouldBe(new[] { "ABC", "", "z" }); } [Fact] public async Task DateTimeArray_PublishesUtcDateTimes() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); var bytes = new byte[8]; BinaryPrimitives.WriteInt64LittleEndian(bytes, 1256102875335L); Birth(ing, TahuDataType.DateTimeArray, bytes); var value = seen.ShouldHaveSingleItem().Value.ShouldBeOfType().ShouldHaveSingleItem(); value.ShouldBe(DateTimeOffset.FromUnixTimeMilliseconds(1256102875335L).UtcDateTime); value.Kind.ShouldBe(DateTimeKind.Utc); } // ---- a declared arrayLength is the node's ArrayDimensions ---- [Fact] public async Task AMatchingDeclaredLength_IsServed() { var (ing, seen) = await SubscribedAsync(ArrayBlob(arrayLength: 3)); Birth(ing, TahuDataType.Int32Array, Int32s(1, 2, 3)); seen.ShouldHaveSingleItem().StatusCode.ShouldBe(Good); } /// A vector that is not the node's declared length is not a value of that node (MTConnect's #627 verdict). [Fact] public async Task AVectorOfAnotherLength_IsBadDecodingError_NotServedUnderTheWrongDimensions() { var (ing, seen) = await SubscribedAsync(ArrayBlob(arrayLength: 4)); Birth(ing, TahuDataType.Int32Array, Int32s(1, 2, 3)); var snapshot = seen.ShouldHaveSingleItem(); snapshot.StatusCode.ShouldBe(BadDecodingError); snapshot.Value.ShouldBeNull(); } // ---- refusals: all-or-nothing, and never the packed buffer ---- [Fact] public async Task AMalformedBody_IsBadDecodingError() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Birth(ing, TahuDataType.Int32Array, [1, 2, 3, 4, 5]); var snapshot = seen.ShouldHaveSingleItem(); snapshot.StatusCode.ShouldBe(BadDecodingError); snapshot.Value.ShouldBeNull(); } /// One element that does not fit the effective type refuses the WHOLE vector — never a partial array. [Fact] public async Task OneElementThatDoesNotFit_RefusesTheWholeVector() { var (ing, seen) = await SubscribedAsync(ArrayBlob(dataType: "Int16")); Birth(ing, TahuDataType.Int32Array, Int32s(1, 70000)); var snapshot = seen.ShouldHaveSingleItem(); snapshot.StatusCode.ShouldBe(BadTypeMismatch); snapshot.Value.ShouldBeNull(); } /// A SCALAR metric on an array tag: the shapes disagree, refused rather than wrapped in a one-element array. [Fact] public async Task AScalarMetric_OnAnArrayTag_IsBadTypeMismatch() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Nbirth(ing); var birth = new Payload { Seq = 1UL, Timestamp = 1UL }; birth.Metrics.Add(new Payload.Types.Metric { Name = "Profile", Alias = ArrAlias, Datatype = (uint)TahuDataType.Int32, IntValue = 5, }); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth)); var snapshot = seen.ShouldHaveSingleItem(); snapshot.StatusCode.ShouldBe(BadTypeMismatch); snapshot.Value.ShouldBeNull(); } /// A publisher that puts a scalar oneof on an array-typed metric: there is no packed body to decode. [Fact] public async Task ANonBytesValue_OnAnArrayMetric_IsBadTypeMismatch() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Nbirth(ing); var birth = new Payload { Seq = 1UL, Timestamp = 1UL }; birth.Metrics.Add(new Payload.Types.Metric { Name = "Profile", Alias = ArrAlias, Datatype = (uint)TahuDataType.Int32Array, IntValue = 5, }); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth)); seen.ShouldHaveSingleItem().StatusCode.ShouldBe(BadTypeMismatch); } [Fact] public async Task AnExplicitNull_OnAnArrayMetric_IsBadTypeMismatch() { var (ing, seen) = await SubscribedAsync(ArrayBlob()); Nbirth(ing); var birth = new Payload { Seq = 1UL, Timestamp = 1UL }; birth.Metrics.Add(new Payload.Types.Metric { Name = "Profile", Alias = ArrAlias, Datatype = (uint)TahuDataType.Int32Array, IsNull = true, }); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth)); seen.ShouldHaveSingleItem().StatusCode.ShouldBe(BadTypeMismatch); } /// /// A template member that is itself an array re-enters the same publish funnel, so it is decoded too — /// the structured path cannot silently lag the top-level one. /// [Fact] public async Task AnArrayTemplateMember_IsDecodedLikeATopLevelArray() { const string MemberPath = "Plant/Mqtt/spb/MotorProfile"; var blob = $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"Motor/Profile","isArray":true}"""; var ing = NewIngestor([new RawTagEntry(MemberPath, blob, WriteIdempotent: false)]); await ing.SubscribeAsync([MemberPath], TimeSpan.Zero, TestContext.Current.CancellationToken); var seen = new List(); ing.OnDataChange += (_, e) => seen.Add(e.Snapshot); Nbirth(ing); var template = new Payload.Types.Template { IsDefinition = false }; template.Metrics.Add(new Payload.Types.Metric { Name = "Profile", Datatype = (uint)TahuDataType.Int16Array, BytesValue = ByteString.CopyFrom(0x01, 0x00, 0xFF, 0xFF), }); var birth = new Payload { Seq = 1UL, Timestamp = 1UL }; birth.Metrics.Add(new Payload.Types.Metric { Name = "Motor", Alias = 30UL, Datatype = (uint)TahuDataType.Template, TemplateValue = template, }); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth)); var snapshot = seen.ShouldHaveSingleItem(); snapshot.StatusCode.ShouldBe(Good); snapshot.Value.ShouldBeOfType().ShouldBe(new short[] { 1, -1 }); } // ---- helpers ---- private static string ArrayBlob(string? dataType = null, uint? arrayLength = null) { var typeKey = dataType is null ? "" : $""","dataType":"{dataType}" """; var lengthKey = arrayLength is null ? "" : $""","arrayLength":{arrayLength}"""; return $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"Profile","isArray":true{{typeKey}}{{lengthKey}}}"""; } private static async Task<(SparkplugIngestor Ingestor, List Seen)> SubscribedAsync(string blob) { var ing = NewIngestor([new RawTagEntry(ArrPath, blob, WriteIdempotent: false)]); await ing.SubscribeAsync([ArrPath], TimeSpan.Zero, TestContext.Current.CancellationToken); var seen = new List(); ing.OnDataChange += (_, e) => seen.Add(e.Snapshot); return (ing, seen); } private static SparkplugIngestor NewIngestor(RawTagEntry[] tags) { var ingestor = new SparkplugIngestor( new MqttDriverOptions { Mode = MqttMode.SparkplugB, Sparkplug = new MqttSparkplugOptions { GroupId = Group, HostId = "otopcua-host-1" }, }, "mqtt-spb-1", null, null, logger: null, cache: null, maxPayloadBytes: MqttSubscriptionManager.DefaultMaxPayloadBytes, rebirthDebounce: TimeSpan.Zero); ingestor.Register(tags); return ingestor; } private static void Nbirth(SparkplugIngestor ing) { var payload = new Payload { Seq = 0UL, Timestamp = 1721822400000UL }; payload.Metrics.Add(new Payload.Types.Metric { Name = "bdSeq", Datatype = (uint)TahuDataType.Uint64, LongValue = 1UL }); ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Decode(payload)); } /// NBIRTH, then a DBIRTH declaring Profile as carrying . private static void Birth(SparkplugIngestor ing, TahuDataType type, byte[] packed) { Nbirth(ing); var birth = new Payload { Seq = 1UL, Timestamp = 1721822400000UL }; birth.Metrics.Add(new Payload.Types.Metric { Name = "Profile", Alias = ArrAlias, Datatype = (uint)type, BytesValue = ByteString.CopyFrom(packed), }); ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth)); } private static byte[] Int32s(params int[] values) { var bytes = new byte[values.Length * 4]; for (var i = 0; i < values.Length; i++) { BinaryPrimitives.WriteInt32LittleEndian(bytes.AsSpan(i * 4), values[i]); } return bytes; } private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) => new(type, Group, Node, device, null); /// Through the wire: encode the generated message, then decode it with the production codec. private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray()); }