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());
}