feat(mqtt): read Sparkplug B array metrics as 1-D OPC UA arrays (lmxopcua#754)
v2-ci / unit-tests (push) Has been cancelled
v2-ci / unit-tests (push) Has been cancelled
R06:M-5 made Sparkplug *Array leaves unbindable because the ingestor refused them. The ingestor now decodes the packed bytes_value of all 13 array types (new SparkplugArrayCodec: LE fixed widths, int64-ms DateTime, uint32-count + MSB-first BooleanArray, NUL-terminated UTF-8 StringArray; golden vectors are the spec's examples, reproduced byte-for-byte by SparkplugNet 1.3.10) and publishes a typed 1-D array through the scalar coercion, all-or-nothing. Shape uses the fleet's one array model: the tag's platform isArray/arrayLength keys (ValueRank=1, ArrayDimensions=[arrayLength] or [0] = variable length). The driver reads them with TagConfigIntent's exact semantics and never rejects on them, so composer, driver, authoring model and deploy inspector agree (MqttArrayShapeParityTests). Shape mismatch (array metric on a scalar tag, or the reverse) is BadTypeMismatch - a String tag no longer gets the packed buffer as base64; a malformed body or a length differing from a declared arrayLength is BadDecodingError (MTConnect #627 verdict). Template/DataSet members of an array type take the same path. Arrays stay read-only (encoder refuses them; MQTT is not in DriverArrayWriteSupport). Browser: array leaves are bindable again and state isArray in AddressFields; the commit mapper writes "isArray": true (no arrayLength). Discovery reports IsArray/ArrayDim. Docs: Mqtt.md "Array metrics", design table, drivers rule. Tests: codec (golden + malformed), ingest through the real codec + Dispatch (the project's MiniBroker never forwards a PUBLISH, so there is no in-process broker harness), browse, discovery, commit mapper, parity corpus. Mutating the fix back out fails 18 MQTT + 9 AdminUI tests. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
@@ -780,12 +780,27 @@ public sealed class MqttBrowseSessionTests
|
||||
a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.Int32)); // element type
|
||||
a.IsArray.ShouldBeTrue();
|
||||
|
||||
// R06:M-5: the ingestor refuses array metrics ("not supported in v1"), so the browser must not
|
||||
// offer one as bindable — no stated address means the browse-commit refuses it in words, instead
|
||||
// of committing a tag that sits at BadWaitingForInitialData forever. The scalar control is
|
||||
// SparkplugAttributes_StateTheBindingTuple_SoTheCommitNeverParsesTheNodeId.
|
||||
a.AddressFields.ShouldBeNull();
|
||||
a.SecurityClass.ShouldContain("not bindable");
|
||||
// #754: the ingestor now reads array metrics as 1-D arrays, so the leaf is bindable again (R06:M-5
|
||||
// made it unbindable while it was not) — but ONLY to an array tag, so the stated address carries the
|
||||
// array bit and the commit writes "isArray": true. Without it the committed tag would be scalar and
|
||||
// read BadTypeMismatch forever, which is R06:M-5 by another route.
|
||||
a.SecurityClass.ShouldNotContain("not bindable");
|
||||
var fields = a.AddressFields.ShouldNotBeNull();
|
||||
fields[MqttTagConfigKeys.MetricName].ShouldBe("Profile");
|
||||
fields[MqttTagConfigKeys.IsArray].ShouldBe("true");
|
||||
}
|
||||
|
||||
/// <summary>Control: a SCALAR metric's address never states the array bit (absence is the scalar default).</summary>
|
||||
[Fact]
|
||||
public async Task SparkplugAttributes_ScalarMetric_DoesNotStateIsArray()
|
||||
{
|
||||
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
||||
s.ObserveBirthForTest("Plant1", "EdgeA", null, "Speed", SparkplugDataType.Int32);
|
||||
|
||||
var a = (await s.AttributesAsync(
|
||||
"Plant1/EdgeA::Speed", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
||||
a.IsArray.ShouldBeFalse();
|
||||
a.AddressFields.ShouldNotBeNull().ContainsKey(MqttTagConfigKeys.IsArray).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
||||
@@ -540,6 +540,36 @@ public sealed class MqttDriverDiscoveryTests
|
||||
b.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.Int32);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #754: an array tag is discovered as what the composer materialises — the birth's ELEMENT type, the
|
||||
/// array bit, and the declared length as <c>ArrayDim</c> (null = variable length). A scalar tag stays
|
||||
/// scalar even when the metric is an array (the ingest refuses that shape; discovery must not hide it).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task DiscoverAsync_Sparkplug_ArrayTag_ReportsElementType_IsArray_AndDeclaredLength()
|
||||
{
|
||||
const string fixedPath = "Plant/Mqtt/spb/Fixed";
|
||||
const string openPath = "Plant/Mqtt/spb/Open";
|
||||
var fixedBlob = SpBlob(SpDevice, "Profile").TrimEnd('}') + ""","isArray":true,"arrayLength":4}""";
|
||||
var openBlob = SpBlob(SpDevice, "Profile").TrimEnd('}') + ""","isArray":true}""";
|
||||
var driver = SparkplugDriver(
|
||||
SpTag(fixedPath, fixedBlob), SpTag(openPath, openBlob), SpTag(SpTempPath, SpBlob(SpDevice, "Profile")));
|
||||
FeedNodeBirth(driver);
|
||||
FeedDeviceBirth(driver, seq: 1, ("Profile", 5UL, TahuDataType.Int32Array));
|
||||
|
||||
var b = new CapturingAddressSpaceBuilder();
|
||||
await driver.DiscoverAsync(b, TestContext.Current.CancellationToken);
|
||||
|
||||
var byName = b.Variables.ToDictionary(v => v.BrowseName, v => v.Info);
|
||||
byName[fixedPath].DriverDataType.ShouldBe(DriverDataType.Int32);
|
||||
byName[fixedPath].IsArray.ShouldBeTrue();
|
||||
byName[fixedPath].ArrayDim.ShouldBe(4u);
|
||||
byName[openPath].IsArray.ShouldBeTrue();
|
||||
byName[openPath].ArrayDim.ShouldBeNull();
|
||||
byName[SpTempPath].IsArray.ShouldBeFalse();
|
||||
byName[SpTempPath].ArrayDim.ShouldBeNull();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A birth declaring a Sparkplug type this driver cannot map (DataSet / Template / PropertySet
|
||||
/// / Unknown) must not blank the tag's type — <c>ToDriverDataType()</c> returns null for those,
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
using System.Buffers.Binary;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
|
||||
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Covers <see cref="SparkplugArrayCodec"/> (lmxopcua#754). The golden vectors are the Sparkplug B 3.0
|
||||
/// specification's own "Datatype details" examples, and an independent encoder (SparkplugNet 1.3.10,
|
||||
/// <c>PayloadConverter.ConvertVersionBMetric</c>) was run against the same inputs and produced the same bytes
|
||||
/// — so the bit order of a BooleanArray and the string terminator are not this repo agreeing with itself.
|
||||
/// </summary>
|
||||
public sealed class SparkplugArrayCodecTests
|
||||
{
|
||||
/// <summary>Decodes and pins the WIRE element type the codec hands the ingestor.</summary>
|
||||
private static T[] Decode<T>(TahuDataType type, params byte[] packed)
|
||||
{
|
||||
SparkplugArrayCodec.TryDecode(type, packed, out var elements, out var error).ShouldBeTrue(error);
|
||||
return elements.ShouldBeOfType<T[]>();
|
||||
}
|
||||
|
||||
private static string Refuse(TahuDataType type, params byte[] packed)
|
||||
{
|
||||
SparkplugArrayCodec.TryDecode(type, packed, out var elements, out var error).ShouldBeFalse();
|
||||
elements.ShouldBeNull();
|
||||
return error.ShouldNotBeNull();
|
||||
}
|
||||
|
||||
// ---- golden vectors (spec examples; SparkplugNet reproduces them byte for byte) ----
|
||||
|
||||
/// <summary>uint32 LE count, then the bits MSB-first: 0x34 = F,F,T,T,F,T,F,F and 0xD0 = T,T,F,T + padding.</summary>
|
||||
[Fact]
|
||||
public void BooleanArray_IsCountPrefixed_AndPackedMostSignificantBitFirst()
|
||||
=> Decode<bool>(TahuDataType.BooleanArray, 0x0C, 0x00, 0x00, 0x00, 0x34, 0xD0)
|
||||
.ShouldBe(new[] { false, false, true, true, false, true, false, false, true, true, false, true });
|
||||
|
||||
[Fact]
|
||||
public void Int8Array_IsOneSignedBytePerElement()
|
||||
=> Decode<sbyte>(TahuDataType.Int8Array, 0xE9, 0x7B).ShouldBe(new sbyte[] { -23, 123 });
|
||||
|
||||
[Fact]
|
||||
public void Int16Array_IsLittleEndian()
|
||||
=> Decode<short>(TahuDataType.Int16Array, 0xD0, 0x8A, 0x30, 0x75).ShouldBe(new short[] { -30000, 30000 });
|
||||
|
||||
[Fact]
|
||||
public void Uint32Array_IsLittleEndian()
|
||||
=> Decode<uint>(TahuDataType.Uint32Array, 0x34, 0x00, 0x00, 0x00, 0x49, 0xFB, 0x55, 0xC4)
|
||||
.ShouldBe(new uint[] { 52, 3293969225 });
|
||||
|
||||
[Fact]
|
||||
public void FloatArray_IsLittleEndianIeee754()
|
||||
=> Decode<float>(TahuDataType.FloatArray, 0xA4, 0x70, 0x9D, 0x3F, 0x98, 0xAE, 0xB2, 0x42)
|
||||
.ShouldBe(new[] { 1.23f, 89.341f });
|
||||
|
||||
[Fact]
|
||||
public void StringArray_IsNullTerminatedUtf8()
|
||||
=> Decode<string>(TahuDataType.StringArray, 0x41, 0x42, 0x43, 0x00, 0x68, 0x65, 0x6C, 0x6C, 0x6F, 0x00)
|
||||
.ShouldBe(new[] { "ABC", "hello" });
|
||||
|
||||
/// <summary>Signed int64 epoch milliseconds — left as the wire long; the ingestor coerces it to a DateTime.</summary>
|
||||
[Fact]
|
||||
public void DateTimeArray_IsInt64EpochMilliseconds()
|
||||
=> Decode<long>(TahuDataType.DateTimeArray, 0xC7, 0xD0, 0x90, 0x75, 0x24, 0x01, 0x00, 0x00)
|
||||
.ShouldBe(new[] { 1256102875335L });
|
||||
|
||||
// ---- the remaining widths, encoded with the BCL so the decoder is not checked against itself ----
|
||||
|
||||
[Fact]
|
||||
public void RemainingWidths_RoundTripTheBclLittleEndianEncoding()
|
||||
{
|
||||
var bytes = new byte[16];
|
||||
BinaryPrimitives.WriteInt64LittleEndian(bytes, -5L);
|
||||
BinaryPrimitives.WriteInt64LittleEndian(bytes.AsSpan(8), long.MaxValue);
|
||||
Decode<long>(TahuDataType.Int64Array, bytes).ShouldBe(new[] { -5L, long.MaxValue });
|
||||
|
||||
BinaryPrimitives.WriteUInt64LittleEndian(bytes, ulong.MaxValue);
|
||||
BinaryPrimitives.WriteUInt64LittleEndian(bytes.AsSpan(8), 7UL);
|
||||
Decode<ulong>(TahuDataType.Uint64Array, bytes).ShouldBe(new[] { ulong.MaxValue, 7UL });
|
||||
|
||||
BinaryPrimitives.WriteDoubleLittleEndian(bytes, -1.5);
|
||||
BinaryPrimitives.WriteDoubleLittleEndian(bytes.AsSpan(8), 1e300);
|
||||
Decode<double>(TahuDataType.DoubleArray, bytes).ShouldBe(new[] { -1.5, 1e300 });
|
||||
|
||||
Decode<byte>(TahuDataType.Uint8Array, 0x00, 0xFF).ShouldBe(new byte[] { 0, 255 });
|
||||
Decode<ushort>(TahuDataType.Uint16Array, 0xFF, 0xFF, 0x01, 0x00).ShouldBe(new ushort[] { 65535, 1 });
|
||||
Decode<int>(TahuDataType.Int32Array, 0xD6, 0xFF, 0xFF, 0xFF).ShouldBe(new[] { -42 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void EmptyBodies_AreEmptyArrays_ExceptBooleanWhichNeedsItsCount()
|
||||
{
|
||||
Decode<int>(TahuDataType.Int32Array).Length.ShouldBe(0);
|
||||
Decode<string>(TahuDataType.StringArray).Length.ShouldBe(0);
|
||||
Decode<bool>(TahuDataType.BooleanArray, 0x00, 0x00, 0x00, 0x00).Length.ShouldBe(0);
|
||||
Refuse(TahuDataType.BooleanArray);
|
||||
}
|
||||
|
||||
// ---- malformed bodies: all-or-nothing, never a partial vector ----
|
||||
|
||||
[Fact]
|
||||
public void ABodyThatIsNotAWholeNumberOfElements_IsRefused()
|
||||
=> Refuse(TahuDataType.Int32Array, 1, 2, 3, 4, 5).ShouldContain("whole number");
|
||||
|
||||
[Fact]
|
||||
public void ABooleanCountTheBodyCannotHold_IsRefused()
|
||||
=> Refuse(TahuDataType.BooleanArray, 0x0C, 0x00, 0x00, 0x00, 0x34).ShouldContain("needs 2");
|
||||
|
||||
/// <summary>Exact, not "at least": a trailing byte means the count and the body disagree.</summary>
|
||||
[Fact]
|
||||
public void ABooleanBodyLongerThanItsCountNeeds_IsRefused()
|
||||
=> Refuse(TahuDataType.BooleanArray, 0x08, 0x00, 0x00, 0x00, 0xFF, 0x00).ShouldContain("needs 1");
|
||||
|
||||
[Fact]
|
||||
public void AnUnterminatedLastString_IsRefused()
|
||||
=> Refuse(TahuDataType.StringArray, 0x41, 0x00, 0x42).ShouldContain("terminated");
|
||||
|
||||
[Fact]
|
||||
public void InvalidUtf8_IsRefused_NotReplacementCharactered()
|
||||
=> Refuse(TahuDataType.StringArray, 0xC0, 0xC0, 0x00).ShouldContain("UTF-8");
|
||||
|
||||
[Fact]
|
||||
public void AScalarDatatype_IsNotAnArray()
|
||||
=> Refuse(TahuDataType.Int32, 0, 0, 0, 0).ShouldContain("not a Sparkplug array");
|
||||
}
|
||||
@@ -0,0 +1,327 @@
|
||||
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;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#754 (R06:M-5): Sparkplug B <c>*Array</c> 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
|
||||
/// <see cref="SparkplugCodec"/> and feeds <see cref="SparkplugIngestor.Dispatch"/> — 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 <c>MiniBroker</c> never forwards a PUBLISH, so there is no in-process broker to route
|
||||
/// through; the live Mosquitto + simulator suite is <c>Driver.Mqtt.IntegrationTests</c>.)
|
||||
/// </summary>
|
||||
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 ----
|
||||
|
||||
/// <summary>An unauthored dataType takes the birth's ELEMENT type, and the value is an int[] — not a byte[].</summary>
|
||||
[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<int[]>().ShouldBe(new[] { -42, 0, 70000 });
|
||||
}
|
||||
|
||||
/// <summary>The DATA path (alias only, no datatype on the wire) decodes by the birth's binding, like a scalar.</summary>
|
||||
[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 });
|
||||
}
|
||||
|
||||
/// <summary>An authored element type wins over the birth's, element by element through the scalar coercion.</summary>
|
||||
[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<double[]>().ShouldBe(new[] { -1d, 2d });
|
||||
}
|
||||
|
||||
/// <summary>Int8 widens to Int16, exactly as the scalar map does (no OPC UA SByte in DriverDataType).</summary>
|
||||
[Fact]
|
||||
public async Task Int8Array_WidensToAnInt16Array()
|
||||
{
|
||||
var (ing, seen) = await SubscribedAsync(ArrayBlob());
|
||||
|
||||
Birth(ing, TahuDataType.Int8Array, [0xE9, 0x7B]);
|
||||
|
||||
seen.ShouldHaveSingleItem().Value.ShouldBeOfType<short[]>().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<bool[]>().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<string[]>().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<DateTime[]>().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);
|
||||
}
|
||||
|
||||
/// <summary>A vector that is not the node's declared length is not a value of that node (MTConnect's #627 verdict).</summary>
|
||||
[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();
|
||||
}
|
||||
|
||||
/// <summary>One element that does not fit the effective type refuses the WHOLE vector — never a partial array.</summary>
|
||||
[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();
|
||||
}
|
||||
|
||||
/// <summary>A SCALAR metric on an array tag: the shapes disagree, refused rather than wrapped in a one-element array.</summary>
|
||||
[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();
|
||||
}
|
||||
|
||||
/// <summary>A publisher that puts a scalar oneof on an array-typed metric: there is no packed body to decode.</summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<DataValueSnapshot>();
|
||||
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<short[]>().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<DataValueSnapshot> 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<DataValueSnapshot>();
|
||||
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));
|
||||
}
|
||||
|
||||
/// <summary>NBIRTH, then a DBIRTH declaring <c>Profile</c> as <paramref name="type"/> carrying <paramref name="packed"/>.</summary>
|
||||
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);
|
||||
|
||||
/// <summary>Through the wire: encode the generated message, then decode it with the production codec.</summary>
|
||||
private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray());
|
||||
}
|
||||
@@ -1344,10 +1344,13 @@ public sealed class SparkplugIngestorTests
|
||||
/// <summary>
|
||||
/// Array metrics slip past the unsupported-type gate, because <c>ToDriverDataType</c> maps every
|
||||
/// <c>*Array</c> variant to its ELEMENT type and is therefore never null for one. The value
|
||||
/// rides as a packed <c>byte[]</c>, so a String-typed tag would publish base64 at GOOD quality.
|
||||
/// rides as a packed <c>byte[]</c>, so a String-typed SCALAR tag would publish base64 at GOOD quality.
|
||||
/// Since #754 an array metric is read — but only onto an array tag; a scalar tag on one is a shape
|
||||
/// mismatch, BadTypeMismatch, never the packed buffer. The array tag itself is covered by
|
||||
/// <see cref="SparkplugArrayIngestTests"/>.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task ArrayMetric_IsSkipped_NotPublishedAsBase64()
|
||||
public async Task ArrayMetric_OnAScalarTag_IsBadTypeMismatch_NotPublishedAsBase64()
|
||||
{
|
||||
const string ArrTag = "Plant/Mqtt/spb/TempArray";
|
||||
var ing = NewIngestor(tags: [Tag(ArrTag, Blob(Group, Node, Device, "Temperature", "String"))]);
|
||||
@@ -1368,8 +1371,10 @@ public sealed class SparkplugIngestorTests
|
||||
});
|
||||
ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth));
|
||||
|
||||
seen.ShouldBeEmpty();
|
||||
ing.Values.Read(ArrTag).StatusCode.ShouldNotBe(Good);
|
||||
var snapshot = seen.ShouldHaveSingleItem();
|
||||
snapshot.StatusCode.ShouldBe(BadTypeMismatch);
|
||||
snapshot.Value.ShouldBeNull();
|
||||
ing.Values.Read(ArrTag).StatusCode.ShouldBe(BadTypeMismatch);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
Reference in New Issue
Block a user